sneak/chat
1
0
Fork 0
Code Issues 15 Pull Requests Actions Packages Projects Releases Wiki Activity

MVP: IRC envelope format, long-polling, per-client queues, SPA rewrite

Browse Source 
Major changes:
- Consolidated schema into single migration with IRC envelope format
- Messages table stores command/from/to/body(JSON)/meta(JSON) per spec
- Per-client delivery queues (client_queues table) with fan-out
- In-memory broker for long-poll notifications (no busy polling)
- GET /messages supports ?after=<queue_id>&timeout=15 long-polling
- All commands (JOIN/PART/NICK/TOPIC/QUIT/PING) broadcast events
- Channels are ephemeral (deleted when last member leaves)
- PRIVMSG to nicks (DMs) fan out to both sender and recipient
- SPA rewritten in vanilla JS (no build step needed):
  - Long-poll via recursive fetch (not setInterval)
  - IRC envelope parsing with system message display
  - /nick, /join, /part, /msg, /quit commands
  - Unread indicators on inactive tabs
  - DM tabs from user list clicks
- Removed unused models package (was for UUID-based schema)
- Removed conflicting UUID-based db methods
- Increased HTTP write timeout to 60s for long-poll support
This commit is contained in:
clawbot
2026-02-10 18:09:10 -08:00
parent df2217a38b
commit 1f54b281fd
20 changed files with 974 additions and 1829 deletions
Download Patch File Download Diff File Expand all files Collapse all files

60
internal/broker/broker.go Normal file
View File

@@ -0,0 +1,60 @@
// Package broker provides an in-memory pub/sub for long-poll notifications.
package broker
import (
"sync"
)
// Broker notifies waiting clients when new messages are available.
type Broker struct {
mu sync.Mutex
listeners map[int64][]chan struct{} // userID -> list of waiting channels
}
// New creates a new Broker.
func New() *Broker {
return &Broker{
listeners: make(map[int64][]chan struct{}),
}
}
// Wait returns a channel that will be closed when a message is available for the user.
func (b *Broker) Wait(userID int64) chan struct{} {
ch := make(chan struct{}, 1)
b.mu.Lock()
b.listeners[userID] = append(b.listeners[userID], ch)
b.mu.Unlock()
return ch
}
// Notify wakes up all waiting clients for a user.
func (b *Broker) Notify(userID int64) {
b.mu.Lock()
waiters := b.listeners[userID]
delete(b.listeners, userID)
b.mu.Unlock()
for _, ch := range waiters {
select {
case ch <- struct{}{}:
default:
}
}
}
// Remove removes a specific wait channel (for cleanup on timeout).
func (b *Broker) Remove(userID int64, ch chan struct{}) {
b.mu.Lock()
defer b.mu.Unlock()
waiters := b.listeners[userID]
for i, w := range waiters {
if w == ch {
b.listeners[userID] = append(waiters[:i], waiters[i+1:]...)
break
}
}
if len(b.listeners[userID]) == 0 {
delete(b.listeners, userID)
}
}

601
internal/db/db.go
View File

@@ -11,11 +11,9 @@ import (
"sort" "sort"
"strconv" "strconv"
"strings" "strings"
"time"
"git.eeqj.de/sneak/chat/internal/config" "git.eeqj.de/sneak/chat/internal/config"
"git.eeqj.de/sneak/chat/internal/logger" "git.eeqj.de/sneak/chat/internal/logger"
"git.eeqj.de/sneak/chat/internal/models"
"go.uber.org/fx" "go.uber.org/fx"
_ "github.com/joho/godotenv/autoload" // loads .env file _ "github.com/joho/godotenv/autoload" // loads .env file
@@ -57,16 +55,13 @@ func New(lc fx.Lifecycle, params Params) (*Database, error) {
lc.Append(fx.Hook{ lc.Append(fx.Hook{
OnStart: func(ctx context.Context) error { OnStart: func(ctx context.Context) error {
s.log.Info("Database OnStart Hook") s.log.Info("Database OnStart Hook")
return s.connect(ctx) return s.connect(ctx)
}, },
OnStop: func(_ context.Context) error { OnStop: func(_ context.Context) error {
s.log.Info("Database OnStop Hook") s.log.Info("Database OnStop Hook")
if s.db != nil { if s.db != nil {
return s.db.Close() return s.db.Close()
} }
return nil return nil
}, },
}) })
@@ -74,455 +69,11 @@ func New(lc fx.Lifecycle, params Params) (*Database, error) {
return s, nil return s, nil
} }
// NewTest creates a Database for testing, bypassing fx lifecycle.
// It connects to the given DSN and runs all migrations.
func NewTest(dsn string) (*Database, error) {
d, err := sql.Open("sqlite", dsn)
if err != nil {
return nil, err
}
s := &Database{
db: d,
log: slog.Default(),
}
// Item 9: Enable foreign keys
if _, err := d.Exec("PRAGMA foreign_keys = ON"); err != nil {
_ = d.Close()
return nil, fmt.Errorf("enable foreign keys: %w", err)
}
ctx := context.Background()
err = s.runMigrations(ctx)
if err != nil {
_ = d.Close()
return nil, err
}
return s, nil
}
// GetDB returns the underlying sql.DB connection. // GetDB returns the underlying sql.DB connection.
func (s *Database) GetDB() *sql.DB { func (s *Database) GetDB() *sql.DB {
return s.db return s.db
} }
// Hydrate injects the database reference into any model that
// embeds Base.
func (s *Database) Hydrate(m interface{ SetDB(d models.DB) }) {
m.SetDB(s)
}
// GetUserByID looks up a user by their ID.
func (s *Database) GetUserByID(
ctx context.Context,
id string,
) (*models.User, error) {
u := &models.User{}
s.Hydrate(u)
err := s.db.QueryRowContext(ctx, `
SELECT id, nick, password_hash, created_at, updated_at, last_seen_at
FROM users WHERE id = ?`,
id,
).Scan(
&u.ID, &u.Nick, &u.PasswordHash,
&u.CreatedAt, &u.UpdatedAt, &u.LastSeenAt,
)
if err != nil {
return nil, err
}
return u, nil
}
// GetChannelByID looks up a channel by its ID.
func (s *Database) GetChannelByID(
ctx context.Context,
id string,
) (*models.Channel, error) {
c := &models.Channel{}
s.Hydrate(c)
err := s.db.QueryRowContext(ctx, `
SELECT id, name, topic, modes, created_at, updated_at
FROM channels WHERE id = ?`,
id,
).Scan(
&c.ID, &c.Name, &c.Topic, &c.Modes,
&c.CreatedAt, &c.UpdatedAt,
)
if err != nil {
return nil, err
}
return c, nil
}
// GetUserByNick looks up a user by their nick.
func (s *Database) GetUserByNick(
ctx context.Context,
nick string,
) (*models.User, error) {
u := &models.User{}
s.Hydrate(u)
err := s.db.QueryRowContext(ctx, `
SELECT id, nick, password_hash, created_at, updated_at, last_seen_at
FROM users WHERE nick = ?`,
nick,
).Scan(
&u.ID, &u.Nick, &u.PasswordHash,
&u.CreatedAt, &u.UpdatedAt, &u.LastSeenAt,
)
if err != nil {
return nil, err
}
return u, nil
}
// GetUserByToken looks up a user by their auth token.
func (s *Database) GetUserByToken(
ctx context.Context,
token string,
) (*models.User, error) {
u := &models.User{}
s.Hydrate(u)
err := s.db.QueryRowContext(ctx, `
SELECT u.id, u.nick, u.password_hash,
u.created_at, u.updated_at, u.last_seen_at
FROM users u
JOIN auth_tokens t ON t.user_id = u.id
WHERE t.token = ?`,
token,
).Scan(
&u.ID, &u.Nick, &u.PasswordHash,
&u.CreatedAt, &u.UpdatedAt, &u.LastSeenAt,
)
if err != nil {
return nil, err
}
return u, nil
}
// DeleteAuthToken removes an auth token from the database.
func (s *Database) DeleteAuthToken(
ctx context.Context,
token string,
) error {
_, err := s.db.ExecContext(ctx,
`DELETE FROM auth_tokens WHERE token = ?`, token,
)
return err
}
// UpdateUserLastSeen updates the last_seen_at timestamp for a user.
func (s *Database) UpdateUserLastSeen(
ctx context.Context,
userID string,
) error {
_, err := s.db.ExecContext(ctx,
`UPDATE users SET last_seen_at = CURRENT_TIMESTAMP WHERE id = ?`,
userID,
)
return err
}
// CreateUser inserts a new user into the database.
func (s *Database) CreateUser(
ctx context.Context,
id, nick, passwordHash string,
) (*models.User, error) {
now := time.Now()
_, err := s.db.ExecContext(ctx,
`INSERT INTO users (id, nick, password_hash)
VALUES (?, ?, ?)`,
id, nick, passwordHash,
)
if err != nil {
return nil, err
}
u := &models.User{
ID: id, Nick: nick, PasswordHash: passwordHash,
CreatedAt: now, UpdatedAt: now,
}
s.Hydrate(u)
return u, nil
}
// CreateChannel inserts a new channel into the database.
func (s *Database) CreateChannel(
ctx context.Context,
id, name, topic, modes string,
) (*models.Channel, error) {
now := time.Now()
_, err := s.db.ExecContext(ctx,
`INSERT INTO channels (id, name, topic, modes)
VALUES (?, ?, ?, ?)`,
id, name, topic, modes,
)
if err != nil {
return nil, err
}
c := &models.Channel{
ID: id, Name: name, Topic: topic, Modes: modes,
CreatedAt: now, UpdatedAt: now,
}
s.Hydrate(c)
return c, nil
}
// AddChannelMember adds a user to a channel with the given modes.
func (s *Database) AddChannelMember(
ctx context.Context,
channelID, userID, modes string,
) (*models.ChannelMember, error) {
now := time.Now()
_, err := s.db.ExecContext(ctx,
`INSERT INTO channel_members
(channel_id, user_id, modes)
VALUES (?, ?, ?)`,
channelID, userID, modes,
)
if err != nil {
return nil, err
}
cm := &models.ChannelMember{
ChannelID: channelID,
UserID: userID,
Modes: modes,
JoinedAt: now,
}
s.Hydrate(cm)
return cm, nil
}
// CreateMessage inserts a new message into the database.
func (s *Database) CreateMessage(
ctx context.Context,
id, fromUserID, fromNick, target, msgType, body string,
) (*models.Message, error) {
now := time.Now()
_, err := s.db.ExecContext(ctx,
`INSERT INTO messages
(id, from_user_id, from_nick, target, type, body)
VALUES (?, ?, ?, ?, ?, ?)`,
id, fromUserID, fromNick, target, msgType, body,
)
if err != nil {
return nil, err
}
m := &models.Message{
ID: id,
FromUserID: fromUserID,
FromNick: fromNick,
Target: target,
Type: msgType,
Body: body,
Timestamp: now,
CreatedAt: now,
}
s.Hydrate(m)
return m, nil
}
// QueueMessage adds a message to a user's delivery queue.
func (s *Database) QueueMessage(
ctx context.Context,
userID, messageID string,
) (*models.MessageQueueEntry, error) {
now := time.Now()
res, err := s.db.ExecContext(ctx,
`INSERT INTO message_queue (user_id, message_id)
VALUES (?, ?)`,
userID, messageID,
)
if err != nil {
return nil, err
}
entryID, err := res.LastInsertId()
if err != nil {
return nil, fmt.Errorf("get last insert id: %w", err)
}
mq := &models.MessageQueueEntry{
ID: entryID,
UserID: userID,
MessageID: messageID,
QueuedAt: now,
}
s.Hydrate(mq)
return mq, nil
}
// DequeueMessages returns up to limit pending messages for a user,
// ordered by queue time (oldest first).
func (s *Database) DequeueMessages(
ctx context.Context,
userID string,
limit int,
) ([]*models.MessageQueueEntry, error) {
rows, err := s.db.QueryContext(ctx, `
SELECT id, user_id, message_id, queued_at
FROM message_queue
WHERE user_id = ?
ORDER BY queued_at ASC
LIMIT ?`,
userID, limit,
)
if err != nil {
return nil, err
}
defer func() { _ = rows.Close() }()
entries := []*models.MessageQueueEntry{}
for rows.Next() {
e := &models.MessageQueueEntry{}
s.Hydrate(e)
err = rows.Scan(&e.ID, &e.UserID, &e.MessageID, &e.QueuedAt)
if err != nil {
return nil, err
}
entries = append(entries, e)
}
return entries, rows.Err()
}
// AckMessages removes the given queue entry IDs, marking them as delivered.
func (s *Database) AckMessages(
ctx context.Context,
entryIDs []int64,
) error {
if len(entryIDs) == 0 {
return nil
}
placeholders := make([]string, len(entryIDs))
args := make([]interface{}, len(entryIDs))
for i, id := range entryIDs {
placeholders[i] = "?"
args[i] = id
}
query := fmt.Sprintf(
"DELETE FROM message_queue WHERE id IN (%s)",
strings.Join(placeholders, ","),
)
_, err := s.db.ExecContext(ctx, query, args...)
return err
}
// CreateAuthToken inserts a new auth token for a user.
func (s *Database) CreateAuthToken(
ctx context.Context,
token, userID string,
) (*models.AuthToken, error) {
now := time.Now()
_, err := s.db.ExecContext(ctx,
`INSERT INTO auth_tokens (token, user_id)
VALUES (?, ?)`,
token, userID,
)
if err != nil {
return nil, err
}
at := &models.AuthToken{Token: token, UserID: userID, CreatedAt: now}
s.Hydrate(at)
return at, nil
}
// CreateSession inserts a new session for a user.
func (s *Database) CreateSession(
ctx context.Context,
id, userID string,
) (*models.Session, error) {
now := time.Now()
_, err := s.db.ExecContext(ctx,
`INSERT INTO sessions (id, user_id)
VALUES (?, ?)`,
id, userID,
)
if err != nil {
return nil, err
}
sess := &models.Session{
ID: id, UserID: userID,
CreatedAt: now, LastActiveAt: now,
}
s.Hydrate(sess)
return sess, nil
}
// CreateServerLink inserts a new server link.
func (s *Database) CreateServerLink(
ctx context.Context,
id, name, url, sharedKeyHash string,
isActive bool,
) (*models.ServerLink, error) {
now := time.Now()
active := 0
if isActive {
active = 1
}
_, err := s.db.ExecContext(ctx,
`INSERT INTO server_links
(id, name, url, shared_key_hash, is_active)
VALUES (?, ?, ?, ?, ?)`,
id, name, url, sharedKeyHash, active,
)
if err != nil {
return nil, err
}
sl := &models.ServerLink{
ID: id,
Name: name,
URL: url,
SharedKeyHash: sharedKeyHash,
IsActive: isActive,
CreatedAt: now,
}
s.Hydrate(sl)
return sl, nil
}
func (s *Database) connect(ctx context.Context) error { func (s *Database) connect(ctx context.Context) error {
dbURL := s.params.Config.DBURL dbURL := s.params.Config.DBURL
if dbURL == "" { if dbURL == "" {
@@ -534,21 +85,18 @@ func (s *Database) connect(ctx context.Context) error {
d, err := sql.Open("sqlite", dbURL) d, err := sql.Open("sqlite", dbURL)
if err != nil { if err != nil {
s.log.Error("failed to open database", "error", err) s.log.Error("failed to open database", "error", err)
return err return err
} }
err = d.PingContext(ctx) err = d.PingContext(ctx)
if err != nil { if err != nil {
s.log.Error("failed to ping database", "error", err) s.log.Error("failed to ping database", "error", err)
return err return err
} }
s.db = d s.db = d
s.log.Info("database connected") s.log.Info("database connected")
// Item 9: Enable foreign keys on every connection
if _, err := s.db.ExecContext(ctx, "PRAGMA foreign_keys = ON"); err != nil { if _, err := s.db.ExecContext(ctx, "PRAGMA foreign_keys = ON"); err != nil {
return fmt.Errorf("enable foreign keys: %w", err) return fmt.Errorf("enable foreign keys: %w", err)
} }
@@ -563,9 +111,13 @@ type migration struct {
} }
func (s *Database) runMigrations(ctx context.Context) error { func (s *Database) runMigrations(ctx context.Context) error {
err := s.bootstrapMigrationsTable(ctx) _, err := s.db.ExecContext(ctx,
`CREATE TABLE IF NOT EXISTS schema_migrations (
version INTEGER PRIMARY KEY,
applied_at DATETIME DEFAULT CURRENT_TIMESTAMP
)`)
if err != nil { if err != nil {
return err return fmt.Errorf("create schema_migrations table: %w", err)
} }
migrations, err := s.loadMigrations() migrations, err := s.loadMigrations()
@@ -573,30 +125,47 @@ func (s *Database) runMigrations(ctx context.Context) error {
return err return err
} }
err = s.applyMigrations(ctx, migrations) for _, m := range migrations {
if err != nil { var exists int
return err err := s.db.QueryRowContext(ctx,
"SELECT COUNT(*) FROM schema_migrations WHERE version = ?",
m.version,
).Scan(&exists)
if err != nil {
return fmt.Errorf("check migration %d: %w", m.version, err)
}
if exists > 0 {
continue
}
s.log.Info("applying migration", "version", m.version, "name", m.name)
tx, err := s.db.BeginTx(ctx, nil)
if err != nil {
return fmt.Errorf("begin tx for migration %d: %w", m.version, err)
}
_, err = tx.ExecContext(ctx, m.sql)
if err != nil {
_ = tx.Rollback()
return fmt.Errorf("apply migration %d (%s): %w", m.version, m.name, err)
}
_, err = tx.ExecContext(ctx,
"INSERT INTO schema_migrations (version) VALUES (?)",
m.version,
)
if err != nil {
_ = tx.Rollback()
return fmt.Errorf("record migration %d: %w", m.version, err)
}
if err := tx.Commit(); err != nil {
return fmt.Errorf("commit migration %d: %w", m.version, err)
}
} }
s.log.Info("database migrations complete") s.log.Info("database migrations complete")
return nil
}
func (s *Database) bootstrapMigrationsTable(
ctx context.Context,
) error {
_, err := s.db.ExecContext(ctx,
`CREATE TABLE IF NOT EXISTS schema_migrations (
version INTEGER PRIMARY KEY,
applied_at DATETIME DEFAULT CURRENT_TIMESTAMP
)`)
if err != nil {
return fmt.Errorf(
"create schema_migrations table: %w", err,
)
}
return nil return nil
} }
@@ -607,16 +176,12 @@ func (s *Database) loadMigrations() ([]migration, error) {
} }
var migrations []migration var migrations []migration
for _, entry := range entries { for _, entry := range entries {
if entry.IsDir() || if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".sql") {
!strings.HasSuffix(entry.Name(), ".sql") {
continue continue
} }
parts := strings.SplitN( parts := strings.SplitN(entry.Name(), "_", minMigrationParts)
entry.Name(), "_", minMigrationParts,
)
if len(parts) < minMigrationParts { if len(parts) < minMigrationParts {
continue continue
} }
@@ -626,13 +191,9 @@ func (s *Database) loadMigrations() ([]migration, error) {
continue continue
} }
content, err := SchemaFiles.ReadFile( content, err := SchemaFiles.ReadFile("schema/" + entry.Name())
"schema/" + entry.Name(),
)
if err != nil { if err != nil {
return nil, fmt.Errorf( return nil, fmt.Errorf("read migration %s: %w", entry.Name(), err)
"read migration %s: %w", entry.Name(), err,
)
} }
migrations = append(migrations, migration{ migrations = append(migrations, migration{
@@ -648,69 +209,3 @@ func (s *Database) loadMigrations() ([]migration, error) {
return migrations, nil return migrations, nil
} }
// Item 4: Wrap each migration in a transaction
func (s *Database) applyMigrations(
ctx context.Context,
migrations []migration,
) error {
for _, m := range migrations {
var exists int
err := s.db.QueryRowContext(ctx,
"SELECT COUNT(*) FROM schema_migrations WHERE version = ?",
m.version,
).Scan(&exists)
if err != nil {
return fmt.Errorf(
"check migration %d: %w", m.version, err,
)
}
if exists > 0 {
continue
}
s.log.Info(
"applying migration",
"version", m.version, "name", m.name,
)
tx, err := s.db.BeginTx(ctx, nil)
if err != nil {
return fmt.Errorf(
"begin tx for migration %d: %w", m.version, err,
)
}
_, err = tx.ExecContext(ctx, m.sql)
if err != nil {
_ = tx.Rollback()
return fmt.Errorf(
"apply migration %d (%s): %w",
m.version, m.name, err,
)
}
_, err = tx.ExecContext(ctx,
"INSERT INTO schema_migrations (version) VALUES (?)",
m.version,
)
if err != nil {
_ = tx.Rollback()
return fmt.Errorf(
"record migration %d: %w", m.version, err,
)
}
if err := tx.Commit(); err != nil {
return fmt.Errorf(
"commit migration %d: %w", m.version, err,
)
}
}
return nil
}

494
internal/db/db_test.go
View File

@@ -1,494 +0,0 @@
package db_test
import (
"context"
"fmt"
"path/filepath"
"testing"
"time"
"git.eeqj.de/sneak/chat/internal/db"
)
const (
nickAlice = "alice"
nickBob = "bob"
nickCharlie = "charlie"
)
// setupTestDB creates a fresh database in a temp directory with
// all migrations applied.
func setupTestDB(t *testing.T) *db.Database {
t.Helper()
dir := t.TempDir()
dsn := fmt.Sprintf(
"file:%s?_journal_mode=WAL",
filepath.Join(dir, "test.db"),
)
d, err := db.NewTest(dsn)
if err != nil {
t.Fatalf("failed to create test database: %v", err)
}
t.Cleanup(func() { _ = d.GetDB().Close() })
return d
}
func TestCreateUser(t *testing.T) {
t.Parallel()
d := setupTestDB(t)
ctx := context.Background()
u, err := d.CreateUser(ctx, "u1", nickAlice, "hash1")
if err != nil {
t.Fatalf("CreateUser: %v", err)
}
if u.ID != "u1" || u.Nick != nickAlice {
t.Errorf("got user %+v", u)
}
}
func TestCreateAuthToken(t *testing.T) {
t.Parallel()
d := setupTestDB(t)
ctx := context.Background()
_, err := d.CreateUser(ctx, "u1", nickAlice, "h")
if err != nil {
t.Fatalf("CreateUser: %v", err)
}
tok, err := d.CreateAuthToken(ctx, "tok1", "u1")
if err != nil {
t.Fatalf("CreateAuthToken: %v", err)
}
if tok.Token != "tok1" || tok.UserID != "u1" {
t.Errorf("unexpected token: %+v", tok)
}
u, err := tok.User(ctx)
if err != nil {
t.Fatalf("AuthToken.User: %v", err)
}
if u.ID != "u1" || u.Nick != nickAlice {
t.Errorf("AuthToken.User got %+v", u)
}
}
func TestCreateChannel(t *testing.T) {
t.Parallel()
d := setupTestDB(t)
ctx := context.Background()
ch, err := d.CreateChannel(
ctx, "c1", "#general", "welcome", "+n",
)
if err != nil {
t.Fatalf("CreateChannel: %v", err)
}
if ch.ID != "c1" || ch.Name != "#general" {
t.Errorf("unexpected channel: %+v", ch)
}
}
func TestAddChannelMember(t *testing.T) {
t.Parallel()
d := setupTestDB(t)
ctx := context.Background()
_, _ = d.CreateUser(ctx, "u1", nickAlice, "h")
_, _ = d.CreateChannel(ctx, "c1", "#general", "", "")
cm, err := d.AddChannelMember(ctx, "c1", "u1", "+o")
if err != nil {
t.Fatalf("AddChannelMember: %v", err)
}
if cm.ChannelID != "c1" || cm.Modes != "+o" {
t.Errorf("unexpected member: %+v", cm)
}
}
func TestCreateMessage(t *testing.T) {
t.Parallel()
d := setupTestDB(t)
ctx := context.Background()
_, _ = d.CreateUser(ctx, "u1", nickAlice, "h")
msg, err := d.CreateMessage(
ctx, "m1", "u1", nickAlice,
"#general", "message", "hello",
)
if err != nil {
t.Fatalf("CreateMessage: %v", err)
}
if msg.ID != "m1" || msg.Body != "hello" {
t.Errorf("unexpected message: %+v", msg)
}
}
func TestQueueMessage(t *testing.T) {
t.Parallel()
d := setupTestDB(t)
ctx := context.Background()
_, _ = d.CreateUser(ctx, "u1", nickAlice, "h")
_, _ = d.CreateUser(ctx, "u2", nickBob, "h")
_, _ = d.CreateMessage(
ctx, "m1", "u1", nickAlice, "u2", "message", "hi",
)
mq, err := d.QueueMessage(ctx, "u2", "m1")
if err != nil {
t.Fatalf("QueueMessage: %v", err)
}
if mq.UserID != "u2" || mq.MessageID != "m1" {
t.Errorf("unexpected queue entry: %+v", mq)
}
}
func TestCreateSession(t *testing.T) {
t.Parallel()
d := setupTestDB(t)
ctx := context.Background()
_, _ = d.CreateUser(ctx, "u1", nickAlice, "h")
sess, err := d.CreateSession(ctx, "s1", "u1")
if err != nil {
t.Fatalf("CreateSession: %v", err)
}
if sess.ID != "s1" || sess.UserID != "u1" {
t.Errorf("unexpected session: %+v", sess)
}
u, err := sess.User(ctx)
if err != nil {
t.Fatalf("Session.User: %v", err)
}
if u.ID != "u1" {
t.Errorf("Session.User got %v, want u1", u.ID)
}
}
func TestCreateServerLink(t *testing.T) {
t.Parallel()
d := setupTestDB(t)
ctx := context.Background()
sl, err := d.CreateServerLink(
ctx, "sl1", "peer1",
"https://peer.example.com", "keyhash", true,
)
if err != nil {
t.Fatalf("CreateServerLink: %v", err)
}
if sl.ID != "sl1" || !sl.IsActive {
t.Errorf("unexpected server link: %+v", sl)
}
}
func TestUserChannels(t *testing.T) {
t.Parallel()
d := setupTestDB(t)
ctx := context.Background()
u, _ := d.CreateUser(ctx, "u1", nickAlice, "h")
_, _ = d.CreateChannel(ctx, "c1", "#alpha", "", "")
_, _ = d.CreateChannel(ctx, "c2", "#beta", "", "")
_, _ = d.AddChannelMember(ctx, "c1", "u1", "")
_, _ = d.AddChannelMember(ctx, "c2", "u1", "")
channels, err := u.Channels(ctx)
if err != nil {
t.Fatalf("User.Channels: %v", err)
}
if len(channels) != 2 {
t.Fatalf("expected 2 channels, got %d", len(channels))
}
if channels[0].Name != "#alpha" {
t.Errorf("first channel: got %s", channels[0].Name)
}
if channels[1].Name != "#beta" {
t.Errorf("second channel: got %s", channels[1].Name)
}
}
func TestUserChannelsEmpty(t *testing.T) {
t.Parallel()
d := setupTestDB(t)
ctx := context.Background()
u, _ := d.CreateUser(ctx, "u1", nickAlice, "h")
channels, err := u.Channels(ctx)
if err != nil {
t.Fatalf("User.Channels: %v", err)
}
if len(channels) != 0 {
t.Errorf("expected 0 channels, got %d", len(channels))
}
}
func TestUserQueuedMessages(t *testing.T) {
t.Parallel()
d := setupTestDB(t)
ctx := context.Background()
u, _ := d.CreateUser(ctx, "u1", nickAlice, "h")
_, _ = d.CreateUser(ctx, "u2", nickBob, "h")
for i := range 3 {
id := fmt.Sprintf("m%d", i)
_, _ = d.CreateMessage(
ctx, id, "u2", nickBob, "u1",
"message", fmt.Sprintf("msg%d", i),
)
time.Sleep(10 * time.Millisecond)
_, _ = d.QueueMessage(ctx, "u1", id)
}
msgs, err := u.QueuedMessages(ctx)
if err != nil {
t.Fatalf("User.QueuedMessages: %v", err)
}
if len(msgs) != 3 {
t.Fatalf("expected 3 messages, got %d", len(msgs))
}
for i, msg := range msgs {
want := fmt.Sprintf("msg%d", i)
if msg.Body != want {
t.Errorf("msg %d: got %q, want %q", i, msg.Body, want)
}
}
}
func TestUserQueuedMessagesEmpty(t *testing.T) {
t.Parallel()
d := setupTestDB(t)
ctx := context.Background()
u, _ := d.CreateUser(ctx, "u1", nickAlice, "h")
msgs, err := u.QueuedMessages(ctx)
if err != nil {
t.Fatalf("User.QueuedMessages: %v", err)
}
if len(msgs) != 0 {
t.Errorf("expected 0 messages, got %d", len(msgs))
}
}
func TestChannelMembers(t *testing.T) {
t.Parallel()
d := setupTestDB(t)
ctx := context.Background()
ch, _ := d.CreateChannel(ctx, "c1", "#general", "", "")
_, _ = d.CreateUser(ctx, "u1", nickAlice, "h")
_, _ = d.CreateUser(ctx, "u2", nickBob, "h")
_, _ = d.CreateUser(ctx, "u3", nickCharlie, "h")
_, _ = d.AddChannelMember(ctx, "c1", "u1", "+o")
_, _ = d.AddChannelMember(ctx, "c1", "u2", "+v")
_, _ = d.AddChannelMember(ctx, "c1", "u3", "")
members, err := ch.Members(ctx)
if err != nil {
t.Fatalf("Channel.Members: %v", err)
}
if len(members) != 3 {
t.Fatalf("expected 3 members, got %d", len(members))
}
nicks := map[string]bool{}
for _, m := range members {
nicks[m.Nick] = true
}
for _, want := range []string{
nickAlice, nickBob, nickCharlie,
} {
if !nicks[want] {
t.Errorf("missing nick %q", want)
}
}
}
func TestChannelMembersEmpty(t *testing.T) {
t.Parallel()
d := setupTestDB(t)
ctx := context.Background()
ch, _ := d.CreateChannel(ctx, "c1", "#empty", "", "")
members, err := ch.Members(ctx)
if err != nil {
t.Fatalf("Channel.Members: %v", err)
}
if len(members) != 0 {
t.Errorf("expected 0, got %d", len(members))
}
}
func TestChannelRecentMessages(t *testing.T) {
t.Parallel()
d := setupTestDB(t)
ctx := context.Background()
ch, _ := d.CreateChannel(ctx, "c1", "#general", "", "")
_, _ = d.CreateUser(ctx, "u1", nickAlice, "h")
for i := range 5 {
id := fmt.Sprintf("m%d", i)
_, _ = d.CreateMessage(
ctx, id, "u1", nickAlice, "#general",
"message", fmt.Sprintf("msg%d", i),
)
time.Sleep(10 * time.Millisecond)
}
msgs, err := ch.RecentMessages(ctx, 3)
if err != nil {
t.Fatalf("RecentMessages: %v", err)
}
if len(msgs) != 3 {
t.Fatalf("expected 3, got %d", len(msgs))
}
if msgs[0].Body != "msg4" {
t.Errorf("first: got %q, want msg4", msgs[0].Body)
}
if msgs[2].Body != "msg2" {
t.Errorf("last: got %q, want msg2", msgs[2].Body)
}
}
func TestChannelRecentMessagesLargeLimit(t *testing.T) {
t.Parallel()
d := setupTestDB(t)
ctx := context.Background()
ch, _ := d.CreateChannel(ctx, "c1", "#general", "", "")
_, _ = d.CreateUser(ctx, "u1", nickAlice, "h")
_, _ = d.CreateMessage(
ctx, "m1", "u1", nickAlice,
"#general", "message", "only",
)
msgs, err := ch.RecentMessages(ctx, 100)
if err != nil {
t.Fatalf("RecentMessages: %v", err)
}
if len(msgs) != 1 {
t.Errorf("expected 1, got %d", len(msgs))
}
}
func TestChannelMemberUser(t *testing.T) {
t.Parallel()
d := setupTestDB(t)
ctx := context.Background()
_, _ = d.CreateUser(ctx, "u1", nickAlice, "h")
_, _ = d.CreateChannel(ctx, "c1", "#general", "", "")
cm, _ := d.AddChannelMember(ctx, "c1", "u1", "+o")
u, err := cm.User(ctx)
if err != nil {
t.Fatalf("ChannelMember.User: %v", err)
}
if u.ID != "u1" || u.Nick != nickAlice {
t.Errorf("got %+v", u)
}
}
func TestChannelMemberChannel(t *testing.T) {
t.Parallel()
d := setupTestDB(t)
ctx := context.Background()
_, _ = d.CreateUser(ctx, "u1", nickAlice, "h")
_, _ = d.CreateChannel(ctx, "c1", "#general", "topic", "+n")
cm, _ := d.AddChannelMember(ctx, "c1", "u1", "")
ch, err := cm.Channel(ctx)
if err != nil {
t.Fatalf("ChannelMember.Channel: %v", err)
}
if ch.ID != "c1" || ch.Topic != "topic" {
t.Errorf("got %+v", ch)
}
}
func TestDMMessage(t *testing.T) {
t.Parallel()
d := setupTestDB(t)
ctx := context.Background()
_, _ = d.CreateUser(ctx, "u1", nickAlice, "h")
_, _ = d.CreateUser(ctx, "u2", nickBob, "h")
msg, err := d.CreateMessage(
ctx, "m1", "u1", nickAlice, "u2", "message", "hey",
)
if err != nil {
t.Fatalf("CreateMessage DM: %v", err)
}
if msg.Target != "u2" {
t.Errorf("target: got %q, want u2", msg.Target)
}
}

353
internal/db/queries.go
View File

@@ -4,8 +4,11 @@ import (
"context" "context"
"crypto/rand" "crypto/rand"
"encoding/hex" "encoding/hex"
"encoding/json"
"fmt" "fmt"
"time" "time"
"github.com/google/uuid"
) )
func generateToken() string { func generateToken() string {
@@ -14,7 +17,34 @@ func generateToken() string {
return hex.EncodeToString(b) return hex.EncodeToString(b)
} }
// CreateUser registers a new user with the given nick and returns the user with token. // IRCMessage is the IRC envelope format for all messages.
type IRCMessage struct {
ID string `json:"id"`
Command string `json:"command"`
From string `json:"from,omitempty"`
To string `json:"to,omitempty"`
Body json.RawMessage `json:"body,omitempty"`
TS string `json:"ts"`
Meta json.RawMessage `json:"meta,omitempty"`
// Internal DB fields (not in JSON)
DBID int64 `json:"-"`
}
// ChannelInfo is a lightweight channel representation.
type ChannelInfo struct {
ID int64 `json:"id"`
Name string `json:"name"`
Topic string `json:"topic"`
}
// MemberInfo represents a channel member.
type MemberInfo struct {
ID int64 `json:"id"`
Nick string `json:"nick"`
LastSeen time.Time `json:"lastSeen"`
}
// CreateUser registers a new user with the given nick.
func (s *Database) CreateUser(ctx context.Context, nick string) (int64, string, error) { func (s *Database) CreateUser(ctx context.Context, nick string) (int64, string, error) {
token := generateToken() token := generateToken()
now := time.Now() now := time.Now()
@@ -36,7 +66,6 @@ func (s *Database) GetUserByToken(ctx context.Context, token string) (int64, str
if err != nil { if err != nil {
return 0, "", err return 0, "", err
} }
// Update last_seen
_, _ = s.db.ExecContext(ctx, "UPDATE users SET last_seen = ? WHERE id = ?", time.Now(), id) _, _ = s.db.ExecContext(ctx, "UPDATE users SET last_seen = ? WHERE id = ?", time.Now(), id)
return id, nick, nil return id, nick, nil
} }
@@ -82,6 +111,15 @@ func (s *Database) PartChannel(ctx context.Context, channelID, userID int64) err
return err return err
} }
// DeleteChannelIfEmpty deletes a channel if it has no members.
func (s *Database) DeleteChannelIfEmpty(ctx context.Context, channelID int64) error {
_, err := s.db.ExecContext(ctx,
`DELETE FROM channels WHERE id = ? AND NOT EXISTS
(SELECT 1 FROM channel_members WHERE channel_id = ?)`,
channelID, channelID)
return err
}
// ListChannels returns all channels the user has joined. // ListChannels returns all channels the user has joined.
func (s *Database) ListChannels(ctx context.Context, userID int64) ([]ChannelInfo, error) { func (s *Database) ListChannels(ctx context.Context, userID int64) ([]ChannelInfo, error) {
rows, err := s.db.QueryContext(ctx, rows, err := s.db.QueryContext(ctx,
@@ -106,11 +144,26 @@ func (s *Database) ListChannels(ctx context.Context, userID int64) ([]ChannelInf
return channels, nil return channels, nil
} }
// ChannelInfo is a lightweight channel representation. // ListAllChannels returns all channels.
type ChannelInfo struct { func (s *Database) ListAllChannels(ctx context.Context) ([]ChannelInfo, error) {
ID int64 `json:"id"` rows, err := s.db.QueryContext(ctx,
Name string `json:"name"` "SELECT id, name, topic FROM channels ORDER BY name")
Topic string `json:"topic"` if err != nil {
return nil, err
}
defer rows.Close()
var channels []ChannelInfo
for rows.Next() {
var ch ChannelInfo
if err := rows.Scan(&ch.ID, &ch.Name, &ch.Topic); err != nil {
return nil, err
}
channels = append(channels, ch)
}
if channels == nil {
channels = []ChannelInfo{}
}
return channels, nil
} }
// ChannelMembers returns all members of a channel. // ChannelMembers returns all members of a channel.
@@ -137,233 +190,155 @@ func (s *Database) ChannelMembers(ctx context.Context, channelID int64) ([]Membe
return members, nil return members, nil
} }
// MemberInfo represents a channel member. // GetChannelMemberIDs returns user IDs of all members in a channel.
type MemberInfo struct { func (s *Database) GetChannelMemberIDs(ctx context.Context, channelID int64) ([]int64, error) {
ID int64 `json:"id"`
Nick string `json:"nick"`
LastSeen time.Time `json:"lastSeen"`
}
// MessageInfo represents a chat message.
type MessageInfo struct {
ID int64 `json:"id"`
Channel string `json:"channel,omitempty"`
Nick string `json:"nick"`
Content string `json:"content"`
IsDM bool `json:"isDm,omitempty"`
DMTarget string `json:"dmTarget,omitempty"`
CreatedAt time.Time `json:"createdAt"`
}
// GetMessages returns messages for a channel, optionally after a given ID.
func (s *Database) GetMessages(ctx context.Context, channelID int64, afterID int64, limit int) ([]MessageInfo, error) {
if limit <= 0 {
limit = 50
}
rows, err := s.db.QueryContext(ctx, rows, err := s.db.QueryContext(ctx,
`SELECT m.id, c.name, u.nick, m.content, m.created_at "SELECT user_id FROM channel_members WHERE channel_id = ?", channelID)
FROM messages m
INNER JOIN users u ON u.id = m.user_id
INNER JOIN channels c ON c.id = m.channel_id
WHERE m.channel_id = ? AND m.is_dm = 0 AND m.id > ?
ORDER BY m.id ASC LIMIT ?`, channelID, afterID, limit)
if err != nil { if err != nil {
return nil, err return nil, err
} }
defer rows.Close() defer rows.Close()
var msgs []MessageInfo var ids []int64
for rows.Next() { for rows.Next() {
var m MessageInfo var id int64
if err := rows.Scan(&m.ID, &m.Channel, &m.Nick, &m.Content, &m.CreatedAt); err != nil { if err := rows.Scan(&id); err != nil {
return nil, err return nil, err
} }
msgs = append(msgs, m) ids = append(ids, id)
} }
if msgs == nil { return ids, nil
msgs = []MessageInfo{}
}
return msgs, nil
} }
// SendMessage inserts a channel message. // GetUserChannelIDs returns channel IDs the user is a member of.
func (s *Database) SendMessage(ctx context.Context, channelID, userID int64, content string) (int64, error) { func (s *Database) GetUserChannelIDs(ctx context.Context, userID int64) ([]int64, error) {
res, err := s.db.ExecContext(ctx,
"INSERT INTO messages (channel_id, user_id, content, is_dm, created_at) VALUES (?, ?, ?, 0, ?)",
channelID, userID, content, time.Now())
if err != nil {
return 0, err
}
return res.LastInsertId()
}
// SendDM inserts a direct message.
func (s *Database) SendDM(ctx context.Context, fromID, toID int64, content string) (int64, error) {
res, err := s.db.ExecContext(ctx,
"INSERT INTO messages (user_id, content, is_dm, dm_target_id, created_at) VALUES (?, ?, 1, ?, ?)",
fromID, content, toID, time.Now())
if err != nil {
return 0, err
}
return res.LastInsertId()
}
// GetDMs returns direct messages between two users after a given ID.
func (s *Database) GetDMs(ctx context.Context, userA, userB int64, afterID int64, limit int) ([]MessageInfo, error) {
if limit <= 0 {
limit = 50
}
rows, err := s.db.QueryContext(ctx, rows, err := s.db.QueryContext(ctx,
`SELECT m.id, u.nick, m.content, t.nick, m.created_at "SELECT channel_id FROM channel_members WHERE user_id = ?", userID)
FROM messages m
INNER JOIN users u ON u.id = m.user_id
INNER JOIN users t ON t.id = m.dm_target_id
WHERE m.is_dm = 1 AND m.id > ?
AND ((m.user_id = ? AND m.dm_target_id = ?) OR (m.user_id = ? AND m.dm_target_id = ?))
ORDER BY m.id ASC LIMIT ?`, afterID, userA, userB, userB, userA, limit)
if err != nil { if err != nil {
return nil, err return nil, err
} }
defer rows.Close() defer rows.Close()
var msgs []MessageInfo var ids []int64
for rows.Next() { for rows.Next() {
var m MessageInfo var id int64
if err := rows.Scan(&m.ID, &m.Nick, &m.Content, &m.DMTarget, &m.CreatedAt); err != nil { if err := rows.Scan(&id); err != nil {
return nil, err return nil, err
} }
m.IsDM = true ids = append(ids, id)
msgs = append(msgs, m)
} }
if msgs == nil { return ids, nil
msgs = []MessageInfo{}
}
return msgs, nil
} }
// PollMessages returns all new messages (channel + DM) for a user after a given ID. // InsertMessage stores a message and returns its DB ID.
func (s *Database) PollMessages(ctx context.Context, userID int64, afterID int64, limit int) ([]MessageInfo, error) { func (s *Database) InsertMessage(ctx context.Context, command, from, to string, body json.RawMessage, meta json.RawMessage) (int64, string, error) {
msgUUID := uuid.New().String()
now := time.Now().UTC()
if body == nil {
body = json.RawMessage("[]")
}
if meta == nil {
meta = json.RawMessage("{}")
}
res, err := s.db.ExecContext(ctx,
`INSERT INTO messages (uuid, command, msg_from, msg_to, body, meta, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?)`,
msgUUID, command, from, to, string(body), string(meta), now)
if err != nil {
return 0, "", err
}
id, _ := res.LastInsertId()
return id, msgUUID, nil
}
// EnqueueMessage adds a message to a user's delivery queue.
func (s *Database) EnqueueMessage(ctx context.Context, userID, messageID int64) error {
_, err := s.db.ExecContext(ctx,
"INSERT OR IGNORE INTO client_queues (user_id, message_id, created_at) VALUES (?, ?, ?)",
userID, messageID, time.Now())
return err
}
// PollMessages returns queued messages for a user after a given queue ID.
func (s *Database) PollMessages(ctx context.Context, userID int64, afterQueueID int64, limit int) ([]IRCMessage, int64, error) {
if limit <= 0 { if limit <= 0 {
limit = 100 limit = 100
} }
rows, err := s.db.QueryContext(ctx, rows, err := s.db.QueryContext(ctx,
`SELECT m.id, COALESCE(c.name, ''), u.nick, m.content, m.is_dm, COALESCE(t.nick, ''), m.created_at `SELECT cq.id, m.uuid, m.command, m.msg_from, m.msg_to, m.body, m.meta, m.created_at
FROM messages m FROM client_queues cq
INNER JOIN users u ON u.id = m.user_id INNER JOIN messages m ON m.id = cq.message_id
LEFT JOIN channels c ON c.id = m.channel_id WHERE cq.user_id = ? AND cq.id > ?
LEFT JOIN users t ON t.id = m.dm_target_id ORDER BY cq.id ASC LIMIT ?`, userID, afterQueueID, limit)
WHERE m.id > ? AND (
(m.is_dm = 0 AND m.channel_id IN (SELECT channel_id FROM channel_members WHERE user_id = ?))
OR (m.is_dm = 1 AND (m.user_id = ? OR m.dm_target_id = ?))
)
ORDER BY m.id ASC LIMIT ?`, afterID, userID, userID, userID, limit)
if err != nil { if err != nil {
return nil, err return nil, afterQueueID, err
} }
defer rows.Close() defer rows.Close()
var msgs []MessageInfo
var msgs []IRCMessage
var lastQID int64
for rows.Next() { for rows.Next() {
var m MessageInfo var m IRCMessage
var isDM int var qID int64
if err := rows.Scan(&m.ID, &m.Channel, &m.Nick, &m.Content, &isDM, &m.DMTarget, &m.CreatedAt); err != nil { var body, meta string
return nil, err var ts time.Time
if err := rows.Scan(&qID, &m.ID, &m.Command, &m.From, &m.To, &body, &meta, &ts); err != nil {
return nil, afterQueueID, err
} }
m.IsDM = isDM == 1 m.Body = json.RawMessage(body)
m.Meta = json.RawMessage(meta)
m.TS = ts.Format(time.RFC3339Nano)
m.DBID = qID
lastQID = qID
msgs = append(msgs, m) msgs = append(msgs, m)
} }
if msgs == nil { if msgs == nil {
msgs = []MessageInfo{} msgs = []IRCMessage{}
} }
return msgs, nil if lastQID == 0 {
lastQID = afterQueueID
}
return msgs, lastQID, nil
} }
// GetMessagesBefore returns channel messages before a given ID (for history scrollback). // GetHistory returns message history for a target (channel or DM nick pair).
func (s *Database) GetMessagesBefore(ctx context.Context, channelID int64, beforeID int64, limit int) ([]MessageInfo, error) { func (s *Database) GetHistory(ctx context.Context, target string, beforeID int64, limit int) ([]IRCMessage, error) {
if limit <= 0 { if limit <= 0 {
limit = 50 limit = 50
} }
var query string var query string
var args []any var args []any
if beforeID > 0 { if beforeID > 0 {
query = `SELECT m.id, c.name, u.nick, m.content, m.created_at query = `SELECT id, uuid, command, msg_from, msg_to, body, meta, created_at
FROM messages m FROM messages WHERE msg_to = ? AND id < ? AND command = 'PRIVMSG'
INNER JOIN users u ON u.id = m.user_id ORDER BY id DESC LIMIT ?`
INNER JOIN channels c ON c.id = m.channel_id args = []any{target, beforeID, limit}
WHERE m.channel_id = ? AND m.is_dm = 0 AND m.id < ?
ORDER BY m.id DESC LIMIT ?`
args = []any{channelID, beforeID, limit}
} else { } else {
query = `SELECT m.id, c.name, u.nick, m.content, m.created_at query = `SELECT id, uuid, command, msg_from, msg_to, body, meta, created_at
FROM messages m FROM messages WHERE msg_to = ? AND command = 'PRIVMSG'
INNER JOIN users u ON u.id = m.user_id ORDER BY id DESC LIMIT ?`
INNER JOIN channels c ON c.id = m.channel_id args = []any{target, limit}
WHERE m.channel_id = ? AND m.is_dm = 0
ORDER BY m.id DESC LIMIT ?`
args = []any{channelID, limit}
} }
rows, err := s.db.QueryContext(ctx, query, args...) rows, err := s.db.QueryContext(ctx, query, args...)
if err != nil { if err != nil {
return nil, err return nil, err
} }
defer rows.Close() defer rows.Close()
var msgs []MessageInfo var msgs []IRCMessage
for rows.Next() { for rows.Next() {
var m MessageInfo var m IRCMessage
if err := rows.Scan(&m.ID, &m.Channel, &m.Nick, &m.Content, &m.CreatedAt); err != nil { var dbID int64
var body, meta string
var ts time.Time
if err := rows.Scan(&dbID, &m.ID, &m.Command, &m.From, &m.To, &body, &meta, &ts); err != nil {
return nil, err return nil, err
} }
m.Body = json.RawMessage(body)
m.Meta = json.RawMessage(meta)
m.TS = ts.Format(time.RFC3339Nano)
m.DBID = dbID
msgs = append(msgs, m) msgs = append(msgs, m)
} }
if msgs == nil { if msgs == nil {
msgs = []MessageInfo{} msgs = []IRCMessage{}
}
// Reverse to ascending order
for i, j := 0, len(msgs)-1; i < j; i, j = i+1, j-1 {
msgs[i], msgs[j] = msgs[j], msgs[i]
}
return msgs, nil
}
// GetDMsBefore returns DMs between two users before a given ID (for history scrollback).
func (s *Database) GetDMsBefore(ctx context.Context, userA, userB int64, beforeID int64, limit int) ([]MessageInfo, error) {
if limit <= 0 {
limit = 50
}
var query string
var args []any
if beforeID > 0 {
query = `SELECT m.id, u.nick, m.content, t.nick, m.created_at
FROM messages m
INNER JOIN users u ON u.id = m.user_id
INNER JOIN users t ON t.id = m.dm_target_id
WHERE m.is_dm = 1 AND m.id < ?
AND ((m.user_id = ? AND m.dm_target_id = ?) OR (m.user_id = ? AND m.dm_target_id = ?))
ORDER BY m.id DESC LIMIT ?`
args = []any{beforeID, userA, userB, userB, userA, limit}
} else {
query = `SELECT m.id, u.nick, m.content, t.nick, m.created_at
FROM messages m
INNER JOIN users u ON u.id = m.user_id
INNER JOIN users t ON t.id = m.dm_target_id
WHERE m.is_dm = 1
AND ((m.user_id = ? AND m.dm_target_id = ?) OR (m.user_id = ? AND m.dm_target_id = ?))
ORDER BY m.id DESC LIMIT ?`
args = []any{userA, userB, userB, userA, limit}
}
rows, err := s.db.QueryContext(ctx, query, args...)
if err != nil {
return nil, err
}
defer rows.Close()
var msgs []MessageInfo
for rows.Next() {
var m MessageInfo
if err := rows.Scan(&m.ID, &m.Nick, &m.Content, &m.DMTarget, &m.CreatedAt); err != nil {
return nil, err
}
m.IsDM = true
msgs = append(msgs, m)
}
if msgs == nil {
msgs = []MessageInfo{}
} }
// Reverse to ascending order // Reverse to ascending order
for i, j := 0, len(msgs)-1; i < j; i, j = i+1, j-1 { for i, j := 0, len(msgs)-1; i < j; i, j = i+1, j-1 {
@@ -380,21 +355,24 @@ func (s *Database) ChangeNick(ctx context.Context, userID int64, newNick string)
} }
// SetTopic sets the topic for a channel. // SetTopic sets the topic for a channel.
func (s *Database) SetTopic(ctx context.Context, channelName string, _ int64, topic string) error { func (s *Database) SetTopic(ctx context.Context, channelName string, topic string) error {
_, err := s.db.ExecContext(ctx, _, err := s.db.ExecContext(ctx,
"UPDATE channels SET topic = ? WHERE name = ?", topic, channelName) "UPDATE channels SET topic = ?, updated_at = ? WHERE name = ?", topic, time.Now(), channelName)
return err return err
} }
// GetServerName returns the server name (unused, config provides this). // DeleteUser removes a user and all their data.
func (s *Database) GetServerName() string { func (s *Database) DeleteUser(ctx context.Context, userID int64) error {
return "" _, err := s.db.ExecContext(ctx, "DELETE FROM users WHERE id = ?", userID)
return err
} }
// ListAllChannels returns all channels. // GetAllChannelMembershipsForUser returns (channelID, channelName) for all channels a user is in.
func (s *Database) ListAllChannels(ctx context.Context) ([]ChannelInfo, error) { func (s *Database) GetAllChannelMembershipsForUser(ctx context.Context, userID int64) ([]ChannelInfo, error) {
rows, err := s.db.QueryContext(ctx, rows, err := s.db.QueryContext(ctx,
"SELECT id, name, topic FROM channels ORDER BY name") `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 { if err != nil {
return nil, err return nil, err
} }
@@ -407,8 +385,5 @@ func (s *Database) ListAllChannels(ctx context.Context) ([]ChannelInfo, error) {
} }
channels = append(channels, ch) channels = append(channels, ch)
} }
if channels == nil {
channels = []ChannelInfo{}
}
return channels, nil return channels, nil
} }

56
internal/db/schema/001_initial.sql
View File

@@ -1,4 +1,54 @@
CREATE TABLE IF NOT EXISTS schema_migrations ( -- Chat server schema (pre-1.0 consolidated)
version INTEGER PRIMARY KEY, PRAGMA foreign_keys = ON;
applied_at DATETIME DEFAULT CURRENT_TIMESTAMP
-- Users: IRC-style sessions (no passwords, just nick + token)
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
nick TEXT NOT NULL UNIQUE,
token TEXT NOT NULL UNIQUE,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
last_seen DATETIME DEFAULT CURRENT_TIMESTAMP
); );
CREATE INDEX IF NOT EXISTS idx_users_token ON users(token);
-- Channels
CREATE TABLE IF NOT EXISTS channels (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL UNIQUE,
topic TEXT NOT NULL DEFAULT '',
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
);
-- Channel members
CREATE TABLE IF NOT EXISTS channel_members (
id INTEGER PRIMARY KEY AUTOINCREMENT,
channel_id INTEGER NOT NULL REFERENCES channels(id) ON DELETE CASCADE,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
joined_at DATETIME DEFAULT CURRENT_TIMESTAMP,
UNIQUE(channel_id, user_id)
);
-- Messages: IRC envelope format
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
uuid TEXT NOT NULL UNIQUE,
command TEXT NOT NULL DEFAULT 'PRIVMSG',
msg_from TEXT NOT NULL DEFAULT '',
msg_to TEXT NOT NULL DEFAULT '',
body TEXT NOT NULL DEFAULT '[]',
meta TEXT NOT NULL DEFAULT '{}',
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX IF NOT EXISTS idx_messages_to_id ON messages(msg_to, id);
CREATE INDEX IF NOT EXISTS idx_messages_created ON messages(created_at);
-- Per-client message queues for fan-out delivery
CREATE TABLE IF NOT EXISTS client_queues (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
message_id INTEGER NOT NULL REFERENCES messages(id) ON DELETE CASCADE,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
UNIQUE(user_id, message_id)
);
CREATE INDEX IF NOT EXISTS idx_client_queues_user ON client_queues(user_id, id);

89
internal/db/schema/002_schema.sql
View File

@@ -1,89 +0,0 @@
-- All schema changes go into this file until 1.0.0 is tagged.
-- There will not be migrations during the early development phase.
-- After 1.0.0, new changes get their own numbered migration files.
-- Users: accounts and authentication
CREATE TABLE IF NOT EXISTS users (
id TEXT PRIMARY KEY, -- UUID
nick TEXT NOT NULL UNIQUE,
password_hash TEXT NOT NULL,
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
last_seen_at DATETIME
);
-- Auth tokens: one user can have multiple active tokens (multiple devices)
CREATE TABLE IF NOT EXISTS auth_tokens (
token TEXT PRIMARY KEY, -- random token string
user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
expires_at DATETIME, -- NULL = no expiry
last_used_at DATETIME
);
CREATE INDEX IF NOT EXISTS idx_auth_tokens_user_id ON auth_tokens(user_id);
-- Channels: chat rooms
CREATE TABLE IF NOT EXISTS channels (
id TEXT PRIMARY KEY, -- UUID
name TEXT NOT NULL UNIQUE, -- #general, etc.
topic TEXT NOT NULL DEFAULT '',
modes TEXT NOT NULL DEFAULT '', -- +i, +m, +s, +t, +n
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
);
-- Channel members: who is in which channel, with per-user modes
CREATE TABLE IF NOT EXISTS channel_members (
channel_id TEXT NOT NULL REFERENCES channels(id) ON DELETE CASCADE,
user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
modes TEXT NOT NULL DEFAULT '', -- +o (operator), +v (voice)
joined_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (channel_id, user_id)
);
CREATE INDEX IF NOT EXISTS idx_channel_members_user_id ON channel_members(user_id);
-- Messages: channel and DM history (rotated per MAX_HISTORY)
CREATE TABLE IF NOT EXISTS messages (
id TEXT PRIMARY KEY, -- UUID
ts DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
from_user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
from_nick TEXT NOT NULL, -- denormalized for history
target TEXT NOT NULL, -- #channel name or user UUID for DMs
type TEXT NOT NULL DEFAULT 'message', -- message, action, notice, join, part, quit, topic, mode, nick, system
body TEXT NOT NULL DEFAULT '',
meta TEXT NOT NULL DEFAULT '{}', -- JSON extensible metadata
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX IF NOT EXISTS idx_messages_target_ts ON messages(target, ts);
CREATE INDEX IF NOT EXISTS idx_messages_from_user ON messages(from_user_id);
-- Message queue: per-user pending delivery (unread messages)
CREATE TABLE IF NOT EXISTS message_queue (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
message_id TEXT NOT NULL REFERENCES messages(id) ON DELETE CASCADE,
queued_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
UNIQUE(user_id, message_id)
);
CREATE INDEX IF NOT EXISTS idx_message_queue_user_id ON message_queue(user_id, queued_at);
-- Sessions: server-held session state
CREATE TABLE IF NOT EXISTS sessions (
id TEXT PRIMARY KEY, -- UUID
user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
last_active_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
expires_at DATETIME -- idle timeout
);
CREATE INDEX IF NOT EXISTS idx_sessions_user_id ON sessions(user_id);
-- Server links: federation peer configuration
CREATE TABLE IF NOT EXISTS server_links (
id TEXT PRIMARY KEY, -- UUID
name TEXT NOT NULL UNIQUE, -- human-readable peer name
url TEXT NOT NULL, -- base URL of peer server
shared_key_hash TEXT NOT NULL, -- hashed shared secret
is_active INTEGER NOT NULL DEFAULT 1,
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
last_seen_at DATETIME
);

31
internal/db/schema/003_users.sql
View File

@@ -1,31 +0,0 @@
PRAGMA foreign_keys = ON;
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
nick TEXT NOT NULL UNIQUE,
token TEXT NOT NULL UNIQUE,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
last_seen DATETIME DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS channel_members (
id INTEGER PRIMARY KEY AUTOINCREMENT,
channel_id INTEGER NOT NULL REFERENCES channels(id) ON DELETE CASCADE,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
joined_at DATETIME DEFAULT CURRENT_TIMESTAMP,
UNIQUE(channel_id, user_id)
);
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
channel_id INTEGER REFERENCES channels(id) ON DELETE CASCADE,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
content TEXT NOT NULL,
is_dm INTEGER NOT NULL DEFAULT 0,
dm_target_id INTEGER REFERENCES users(id) ON DELETE CASCADE,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX IF NOT EXISTS idx_messages_channel ON messages(channel_id, created_at);
CREATE INDEX IF NOT EXISTS idx_messages_dm ON messages(user_id, dm_target_id, created_at);
CREATE INDEX IF NOT EXISTS idx_users_token ON users(token);

258
internal/handlers/api.go
View File

@@ -6,8 +6,8 @@ import (
"net/http" "net/http"
"strconv" "strconv"
"strings" "strings"
"time"
"git.eeqj.de/sneak/chat/internal/db"
"github.com/go-chi/chi" "github.com/go-chi/chi"
) )
@@ -30,6 +30,48 @@ func (s *Handlers) requireAuth(w http.ResponseWriter, r *http.Request) (int64, s
return uid, nick, true return uid, nick, true
} }
// fanOut stores a message and enqueues it to all specified user IDs, then notifies them.
func (s *Handlers) fanOut(ctx *http.Request, command, from, to string, body json.RawMessage, userIDs []int64) error {
dbID, _, err := s.params.Database.InsertMessage(ctx.Context(), command, from, to, body, nil)
if err != nil {
return err
}
for _, uid := range userIDs {
if err := s.params.Database.EnqueueMessage(ctx.Context(), uid, dbID); err != nil {
s.log.Error("enqueue failed", "error", err, "user_id", uid)
}
s.broker.Notify(uid)
}
return nil
}
// fanOutRaw stores and fans out, returning the message DB ID.
func (s *Handlers) fanOutDirect(ctx *http.Request, command, from, to string, body json.RawMessage, userIDs []int64) (int64, string, error) {
dbID, msgUUID, err := s.params.Database.InsertMessage(ctx.Context(), command, from, to, body, nil)
if err != nil {
return 0, "", err
}
for _, uid := range userIDs {
if err := s.params.Database.EnqueueMessage(ctx.Context(), uid, dbID); err != nil {
s.log.Error("enqueue failed", "error", err, "user_id", uid)
}
s.broker.Notify(uid)
}
return dbID, msgUUID, nil
}
// getChannelMembers gets all member IDs for a channel by name.
func (s *Handlers) getChannelMemberIDs(r *http.Request, channelName string) (int64, []int64, error) {
var chID int64
err := s.params.Database.GetDB().QueryRowContext(r.Context(),
"SELECT id FROM channels WHERE name = ?", channelName).Scan(&chID)
if err != nil {
return 0, nil, err
}
ids, err := s.params.Database.GetChannelMemberIDs(r.Context(), chID)
return chID, ids, err
}
// HandleCreateSession creates a new user session and returns the auth token. // HandleCreateSession creates a new user session and returns the auth token.
func (s *Handlers) HandleCreateSession() http.HandlerFunc { func (s *Handlers) HandleCreateSession() http.HandlerFunc {
type request struct { type request struct {
@@ -67,11 +109,6 @@ func (s *Handlers) HandleCreateSession() http.HandlerFunc {
// HandleState returns the current user's info and joined channels. // HandleState returns the current user's info and joined channels.
func (s *Handlers) HandleState() http.HandlerFunc { func (s *Handlers) HandleState() http.HandlerFunc {
type response struct {
ID int64 `json:"id"`
Nick string `json:"nick"`
Channels []db.ChannelInfo `json:"channels"`
}
return func(w http.ResponseWriter, r *http.Request) { return func(w http.ResponseWriter, r *http.Request) {
uid, nick, ok := s.requireAuth(w, r) uid, nick, ok := s.requireAuth(w, r)
if !ok { if !ok {
@@ -83,7 +120,11 @@ func (s *Handlers) HandleState() http.HandlerFunc {
s.respondJSON(w, r, map[string]string{"error": "internal error"}, http.StatusInternalServerError) s.respondJSON(w, r, map[string]string{"error": "internal error"}, http.StatusInternalServerError)
return return
} }
s.respondJSON(w, r, &response{ID: uid, Nick: nick, Channels: channels}, http.StatusOK) s.respondJSON(w, r, map[string]any{
"id": uid,
"nick": nick,
"channels": channels,
}, http.StatusOK)
} }
} }
@@ -129,8 +170,7 @@ func (s *Handlers) HandleChannelMembers() http.HandlerFunc {
} }
} }
// HandleGetMessages returns all new messages (channel + DM) for the user via long-polling. // HandleGetMessages returns messages via long-polling from the client's queue.
// This is the single unified message stream — replaces separate channel/DM/poll endpoints.
func (s *Handlers) HandleGetMessages() http.HandlerFunc { func (s *Handlers) HandleGetMessages() http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) { return func(w http.ResponseWriter, r *http.Request) {
uid, _, ok := s.requireAuth(w, r) uid, _, ok := s.requireAuth(w, r)
@@ -138,25 +178,65 @@ func (s *Handlers) HandleGetMessages() http.HandlerFunc {
return return
} }
afterID, _ := strconv.ParseInt(r.URL.Query().Get("after"), 10, 64) afterID, _ := strconv.ParseInt(r.URL.Query().Get("after"), 10, 64)
limit, _ := strconv.Atoi(r.URL.Query().Get("limit")) timeout, _ := strconv.Atoi(r.URL.Query().Get("timeout"))
msgs, err := s.params.Database.PollMessages(r.Context(), uid, afterID, limit) if timeout <= 0 {
timeout = 0
}
if timeout > 30 {
timeout = 30
}
// First check for existing messages.
msgs, lastQID, err := s.params.Database.PollMessages(r.Context(), uid, afterID, 100)
if err != nil { if err != nil {
s.log.Error("get messages failed", "error", err) s.log.Error("poll messages failed", "error", err)
s.respondJSON(w, r, map[string]string{"error": "internal error"}, http.StatusInternalServerError) s.respondJSON(w, r, map[string]string{"error": "internal error"}, http.StatusInternalServerError)
return return
} }
s.respondJSON(w, r, msgs, http.StatusOK)
if len(msgs) > 0 || timeout == 0 {
s.respondJSON(w, r, map[string]any{
"messages": msgs,
"last_id": lastQID,
}, http.StatusOK)
return
}
// Long-poll: wait for notification or timeout.
waitCh := s.broker.Wait(uid)
timer := time.NewTimer(time.Duration(timeout) * time.Second)
defer timer.Stop()
select {
case <-waitCh:
case <-timer.C:
case <-r.Context().Done():
s.broker.Remove(uid, waitCh)
return
}
s.broker.Remove(uid, waitCh)
// Check again after notification.
msgs, lastQID, err = s.params.Database.PollMessages(r.Context(), uid, afterID, 100)
if err != nil {
s.log.Error("poll messages failed", "error", err)
s.respondJSON(w, r, map[string]string{"error": "internal error"}, http.StatusInternalServerError)
return
}
s.respondJSON(w, r, map[string]any{
"messages": msgs,
"last_id": lastQID,
}, http.StatusOK)
} }
} }
// HandleSendCommand handles all C2S commands via POST /messages. // HandleSendCommand handles all C2S commands via POST /messages.
// The "command" field dispatches to the appropriate logic.
func (s *Handlers) HandleSendCommand() http.HandlerFunc { func (s *Handlers) HandleSendCommand() http.HandlerFunc {
type request struct { type request struct {
Command string `json:"command"` Command string `json:"command"`
To string `json:"to"` To string `json:"to"`
Params []string `json:"params,omitempty"` Body json.RawMessage `json:"body,omitempty"`
Body interface{} `json:"body,omitempty"` Meta json.RawMessage `json:"meta,omitempty"`
} }
return func(w http.ResponseWriter, r *http.Request) { return func(w http.ResponseWriter, r *http.Request) {
uid, nick, ok := s.requireAuth(w, r) uid, nick, ok := s.requireAuth(w, r)
@@ -171,22 +251,15 @@ func (s *Handlers) HandleSendCommand() http.HandlerFunc {
req.Command = strings.ToUpper(strings.TrimSpace(req.Command)) req.Command = strings.ToUpper(strings.TrimSpace(req.Command))
req.To = strings.TrimSpace(req.To) req.To = strings.TrimSpace(req.To)
// Helper to extract body as string lines.
bodyLines := func() []string { bodyLines := func() []string {
switch v := req.Body.(type) { if req.Body == nil {
case []interface{}:
lines := make([]string, 0, len(v))
for _, item := range v {
if s, ok := item.(string); ok {
lines = append(lines, s)
}
}
return lines
case []string:
return v
default:
return nil return nil
} }
var lines []string
if err := json.Unmarshal(req.Body, &lines); err != nil {
return nil
}
return lines
} }
switch req.Command { switch req.Command {
@@ -200,38 +273,39 @@ func (s *Handlers) HandleSendCommand() http.HandlerFunc {
s.respondJSON(w, r, map[string]string{"error": "body required"}, http.StatusBadRequest) s.respondJSON(w, r, map[string]string{"error": "body required"}, http.StatusBadRequest)
return return
} }
content := strings.Join(lines, "\n")
if strings.HasPrefix(req.To, "#") { if strings.HasPrefix(req.To, "#") {
// Channel message // Channel message — fan out to all channel members.
var chID int64 _, memberIDs, err := s.getChannelMemberIDs(r, req.To)
err := s.params.Database.GetDB().QueryRowContext(r.Context(),
"SELECT id FROM channels WHERE name = ?", req.To).Scan(&chID)
if err != nil { if err != nil {
s.respondJSON(w, r, map[string]string{"error": "channel not found"}, http.StatusNotFound) s.respondJSON(w, r, map[string]string{"error": "channel not found"}, http.StatusNotFound)
return return
} }
msgID, err := s.params.Database.SendMessage(r.Context(), chID, uid, content) _, msgUUID, err := s.fanOutDirect(r, req.Command, nick, req.To, req.Body, memberIDs)
if err != nil { if err != nil {
s.log.Error("send message failed", "error", err) s.log.Error("send message failed", "error", err)
s.respondJSON(w, r, map[string]string{"error": "internal error"}, http.StatusInternalServerError) s.respondJSON(w, r, map[string]string{"error": "internal error"}, http.StatusInternalServerError)
return return
} }
s.respondJSON(w, r, map[string]any{"id": msgID, "status": "sent"}, http.StatusCreated) s.respondJSON(w, r, map[string]string{"id": msgUUID, "status": "sent"}, http.StatusCreated)
} else { } else {
// DM // DM — fan out to recipient + sender.
targetID, err := s.params.Database.GetUserByNick(r.Context(), req.To) targetUID, err := s.params.Database.GetUserByNick(r.Context(), req.To)
if err != nil { if err != nil {
s.respondJSON(w, r, map[string]string{"error": "user not found"}, http.StatusNotFound) s.respondJSON(w, r, map[string]string{"error": "user not found"}, http.StatusNotFound)
return return
} }
msgID, err := s.params.Database.SendDM(r.Context(), uid, targetID, content) recipients := []int64{targetUID}
if targetUID != uid {
recipients = append(recipients, uid) // echo to sender
}
_, msgUUID, err := s.fanOutDirect(r, req.Command, nick, req.To, req.Body, recipients)
if err != nil { if err != nil {
s.log.Error("send dm failed", "error", err) s.log.Error("send dm failed", "error", err)
s.respondJSON(w, r, map[string]string{"error": "internal error"}, http.StatusInternalServerError) s.respondJSON(w, r, map[string]string{"error": "internal error"}, http.StatusInternalServerError)
return return
} }
s.respondJSON(w, r, map[string]any{"id": msgID, "status": "sent"}, http.StatusCreated) s.respondJSON(w, r, map[string]string{"id": msgUUID, "status": "sent"}, http.StatusCreated)
} }
case "JOIN": case "JOIN":
@@ -254,6 +328,9 @@ func (s *Handlers) HandleSendCommand() http.HandlerFunc {
s.respondJSON(w, r, map[string]string{"error": "internal error"}, http.StatusInternalServerError) s.respondJSON(w, r, map[string]string{"error": "internal error"}, http.StatusInternalServerError)
return return
} }
// Broadcast JOIN to all channel members (including the joiner).
memberIDs, _ := s.params.Database.GetChannelMemberIDs(r.Context(), chID)
_ = s.fanOut(r, "JOIN", nick, channel, nil, memberIDs)
s.respondJSON(w, r, map[string]string{"status": "joined", "channel": channel}, http.StatusOK) s.respondJSON(w, r, map[string]string{"status": "joined", "channel": channel}, http.StatusOK)
case "PART": case "PART":
@@ -272,11 +349,17 @@ func (s *Handlers) HandleSendCommand() http.HandlerFunc {
s.respondJSON(w, r, map[string]string{"error": "channel not found"}, http.StatusNotFound) s.respondJSON(w, r, map[string]string{"error": "channel not found"}, http.StatusNotFound)
return return
} }
// Broadcast PART before removing the member.
memberIDs, _ := s.params.Database.GetChannelMemberIDs(r.Context(), chID)
_ = s.fanOut(r, "PART", nick, channel, req.Body, memberIDs)
if err := s.params.Database.PartChannel(r.Context(), chID, uid); err != nil { if err := s.params.Database.PartChannel(r.Context(), chID, uid); err != nil {
s.log.Error("part channel failed", "error", err) s.log.Error("part channel failed", "error", err)
s.respondJSON(w, r, map[string]string{"error": "internal error"}, http.StatusInternalServerError) s.respondJSON(w, r, map[string]string{"error": "internal error"}, http.StatusInternalServerError)
return return
} }
// Delete channel if empty (ephemeral).
_ = s.params.Database.DeleteChannelIfEmpty(r.Context(), chID)
s.respondJSON(w, r, map[string]string{"status": "parted", "channel": channel}, http.StatusOK) s.respondJSON(w, r, map[string]string{"status": "parted", "channel": channel}, http.StatusOK)
case "NICK": case "NICK":
@@ -299,6 +382,25 @@ func (s *Handlers) HandleSendCommand() http.HandlerFunc {
s.respondJSON(w, r, map[string]string{"error": "internal error"}, http.StatusInternalServerError) s.respondJSON(w, r, map[string]string{"error": "internal error"}, http.StatusInternalServerError)
return return
} }
// Broadcast NICK to all channels the user is in.
channels, _ := s.params.Database.GetAllChannelMembershipsForUser(r.Context(), uid)
notified := map[int64]bool{uid: true}
body, _ := json.Marshal([]string{newNick})
// Notify self.
dbID, _, _ := s.params.Database.InsertMessage(r.Context(), "NICK", nick, "", json.RawMessage(body), nil)
_ = s.params.Database.EnqueueMessage(r.Context(), uid, dbID)
s.broker.Notify(uid)
for _, ch := range channels {
memberIDs, _ := s.params.Database.GetChannelMemberIDs(r.Context(), ch.ID)
for _, mid := range memberIDs {
if !notified[mid] {
notified[mid] = true
_ = s.params.Database.EnqueueMessage(r.Context(), mid, dbID)
s.broker.Notify(mid)
}
}
}
s.respondJSON(w, r, map[string]string{"status": "ok", "nick": newNick}, http.StatusOK) s.respondJSON(w, r, map[string]string{"status": "ok", "nick": newNick}, http.StatusOK)
case "TOPIC": case "TOPIC":
@@ -316,27 +418,52 @@ func (s *Handlers) HandleSendCommand() http.HandlerFunc {
if !strings.HasPrefix(channel, "#") { if !strings.HasPrefix(channel, "#") {
channel = "#" + channel channel = "#" + channel
} }
if err := s.params.Database.SetTopic(r.Context(), channel, uid, topic); err != nil { if err := s.params.Database.SetTopic(r.Context(), channel, topic); err != nil {
s.log.Error("set topic failed", "error", err) s.log.Error("set topic failed", "error", err)
s.respondJSON(w, r, map[string]string{"error": "internal error"}, http.StatusInternalServerError) s.respondJSON(w, r, map[string]string{"error": "internal error"}, http.StatusInternalServerError)
return return
} }
// Broadcast TOPIC to channel members.
_, memberIDs, _ := s.getChannelMemberIDs(r, channel)
_ = s.fanOut(r, "TOPIC", nick, channel, req.Body, memberIDs)
s.respondJSON(w, r, map[string]string{"status": "ok", "topic": topic}, http.StatusOK) s.respondJSON(w, r, map[string]string{"status": "ok", "topic": topic}, http.StatusOK)
case "QUIT":
// Broadcast QUIT to all channels, then remove user.
channels, _ := s.params.Database.GetAllChannelMembershipsForUser(r.Context(), uid)
notified := map[int64]bool{}
var dbID int64
if len(channels) > 0 {
dbID, _, _ = s.params.Database.InsertMessage(r.Context(), "QUIT", nick, "", req.Body, nil)
}
for _, ch := range channels {
memberIDs, _ := s.params.Database.GetChannelMemberIDs(r.Context(), ch.ID)
for _, mid := range memberIDs {
if mid != uid && !notified[mid] {
notified[mid] = true
_ = s.params.Database.EnqueueMessage(r.Context(), mid, dbID)
s.broker.Notify(mid)
}
}
_ = s.params.Database.PartChannel(r.Context(), ch.ID, uid)
_ = s.params.Database.DeleteChannelIfEmpty(r.Context(), ch.ID)
}
_ = s.params.Database.DeleteUser(r.Context(), uid)
s.respondJSON(w, r, map[string]string{"status": "quit"}, http.StatusOK)
case "PING": case "PING":
s.respondJSON(w, r, map[string]string{"command": "PONG", "from": s.params.Config.ServerName}, http.StatusOK) s.respondJSON(w, r, map[string]string{"command": "PONG", "from": s.params.Config.ServerName}, http.StatusOK)
default: default:
_ = nick // suppress unused warning
s.respondJSON(w, r, map[string]string{"error": "unknown command: " + req.Command}, http.StatusBadRequest) s.respondJSON(w, r, map[string]string{"error": "unknown command: " + req.Command}, http.StatusBadRequest)
} }
} }
} }
// HandleGetHistory returns message history for a specific target (channel or DM). // HandleGetHistory returns message history for a specific target.
func (s *Handlers) HandleGetHistory() http.HandlerFunc { func (s *Handlers) HandleGetHistory() http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) { return func(w http.ResponseWriter, r *http.Request) {
uid, _, ok := s.requireAuth(w, r) _, _, ok := s.requireAuth(w, r)
if !ok { if !ok {
return return
} }
@@ -350,42 +477,17 @@ func (s *Handlers) HandleGetHistory() http.HandlerFunc {
if limit <= 0 { if limit <= 0 {
limit = 50 limit = 50
} }
msgs, err := s.params.Database.GetHistory(r.Context(), target, beforeID, limit)
if strings.HasPrefix(target, "#") { if err != nil {
// Channel history s.log.Error("get history failed", "error", err)
var chID int64 s.respondJSON(w, r, map[string]string{"error": "internal error"}, http.StatusInternalServerError)
err := s.params.Database.GetDB().QueryRowContext(r.Context(), return
"SELECT id FROM channels WHERE name = ?", target).Scan(&chID)
if err != nil {
s.respondJSON(w, r, map[string]string{"error": "channel not found"}, http.StatusNotFound)
return
}
msgs, err := s.params.Database.GetMessagesBefore(r.Context(), chID, beforeID, limit)
if err != nil {
s.log.Error("get history failed", "error", err)
s.respondJSON(w, r, map[string]string{"error": "internal error"}, http.StatusInternalServerError)
return
}
s.respondJSON(w, r, msgs, http.StatusOK)
} else {
// DM history
targetID, err := s.params.Database.GetUserByNick(r.Context(), target)
if err != nil {
s.respondJSON(w, r, map[string]string{"error": "user not found"}, http.StatusNotFound)
return
}
msgs, err := s.params.Database.GetDMsBefore(r.Context(), uid, targetID, beforeID, limit)
if err != nil {
s.log.Error("get dm history failed", "error", err)
s.respondJSON(w, r, map[string]string{"error": "internal error"}, http.StatusInternalServerError)
return
}
s.respondJSON(w, r, msgs, http.StatusOK)
} }
s.respondJSON(w, r, msgs, http.StatusOK)
} }
} }
// HandleServerInfo returns server metadata (MOTD, name). // HandleServerInfo returns server metadata.
func (s *Handlers) HandleServerInfo() http.HandlerFunc { func (s *Handlers) HandleServerInfo() http.HandlerFunc {
type response struct { type response struct {
Name string `json:"name"` Name string `json:"name"`

5
internal/handlers/handlers.go
View File

@@ -7,6 +7,7 @@ import (
"log/slog" "log/slog"
"net/http" "net/http"
"git.eeqj.de/sneak/chat/internal/broker"
"git.eeqj.de/sneak/chat/internal/config" "git.eeqj.de/sneak/chat/internal/config"
"git.eeqj.de/sneak/chat/internal/db" "git.eeqj.de/sneak/chat/internal/db"
"git.eeqj.de/sneak/chat/internal/globals" "git.eeqj.de/sneak/chat/internal/globals"
@@ -31,6 +32,7 @@ type Handlers struct {
params *Params params *Params
log *slog.Logger log *slog.Logger
hc *healthcheck.Healthcheck hc *healthcheck.Healthcheck
broker *broker.Broker
} }
// New creates a new Handlers instance. // New creates a new Handlers instance.
@@ -39,6 +41,7 @@ func New(lc fx.Lifecycle, params Params) (*Handlers, error) {
s.params = &params s.params = &params
s.log = params.Logger.Get() s.log = params.Logger.Get()
s.hc = params.Healthcheck s.hc = params.Healthcheck
s.broker = broker.New()
lc.Append(fx.Hook{ lc.Append(fx.Hook{
OnStart: func(_ context.Context) error { OnStart: func(_ context.Context) error {
@@ -50,8 +53,8 @@ func New(lc fx.Lifecycle, params Params) (*Handlers, error) {
} }
func (s *Handlers) respondJSON(w http.ResponseWriter, _ *http.Request, data any, status int) { func (s *Handlers) respondJSON(w http.ResponseWriter, _ *http.Request, data any, status int) {
w.WriteHeader(status)
w.Header().Set("Content-Type", "application/json") w.Header().Set("Content-Type", "application/json")
w.WriteHeader(status)
if data != nil { if data != nil {
err := json.NewEncoder(w).Encode(data) err := json.NewEncoder(w).Encode(data)

27
internal/models/auth_token.go
View File

@@ -1,27 +0,0 @@
package models
import (
"context"
"fmt"
"time"
)
// AuthToken represents an authentication token for a user session.
type AuthToken struct {
Base
Token string `json:"-"`
UserID string `json:"userId"`
CreatedAt time.Time `json:"createdAt"`
ExpiresAt *time.Time `json:"expiresAt,omitempty"`
LastUsedAt *time.Time `json:"lastUsedAt,omitempty"`
}
// User returns the user who owns this token.
func (t *AuthToken) User(ctx context.Context) (*User, error) {
if ul := t.GetUserLookup(); ul != nil {
return ul.GetUserByID(ctx, t.UserID)
}
return nil, fmt.Errorf("user lookup not available")
}

96
internal/models/channel.go
View File

@@ -1,96 +0,0 @@
package models
import (
"context"
"time"
)
// Channel represents a chat channel.
type Channel struct {
Base
ID string `json:"id"`
Name string `json:"name"`
Topic string `json:"topic"`
Modes string `json:"modes"`
CreatedAt time.Time `json:"createdAt"`
UpdatedAt time.Time `json:"updatedAt"`
}
// Members returns all users who are members of this channel.
func (c *Channel) Members(ctx context.Context) ([]*ChannelMember, error) {
rows, err := c.GetDB().QueryContext(ctx, `
SELECT cm.channel_id, cm.user_id, cm.modes, cm.joined_at,
u.nick
FROM channel_members cm
JOIN users u ON u.id = cm.user_id
WHERE cm.channel_id = ?
ORDER BY cm.joined_at`,
c.ID,
)
if err != nil {
return nil, err
}
defer func() { _ = rows.Close() }()
members := []*ChannelMember{}
for rows.Next() {
m := &ChannelMember{}
m.SetDB(c.db)
err = rows.Scan(
&m.ChannelID, &m.UserID, &m.Modes,
&m.JoinedAt, &m.Nick,
)
if err != nil {
return nil, err
}
members = append(members, m)
}
return members, rows.Err()
}
// RecentMessages returns the most recent messages in this channel.
func (c *Channel) RecentMessages(
ctx context.Context,
limit int,
) ([]*Message, error) {
rows, err := c.GetDB().QueryContext(ctx, `
SELECT id, ts, from_user_id, from_nick,
target, type, body, meta, created_at
FROM messages
WHERE target = ?
ORDER BY ts DESC
LIMIT ?`,
c.Name, limit,
)
if err != nil {
return nil, err
}
defer func() { _ = rows.Close() }()
messages := []*Message{}
for rows.Next() {
msg := &Message{}
msg.SetDB(c.db)
err = rows.Scan(
&msg.ID, &msg.Timestamp, &msg.FromUserID,
&msg.FromNick, &msg.Target, &msg.Type,
&msg.Body, &msg.Meta, &msg.CreatedAt,
)
if err != nil {
return nil, err
}
messages = append(messages, msg)
}
return messages, rows.Err()
}

36
internal/models/channel_member.go
View File

@@ -1,36 +0,0 @@
package models
import (
"context"
"fmt"
"time"
)
// ChannelMember represents a user's membership in a channel.
type ChannelMember struct {
Base
ChannelID string `json:"channelId"`
UserID string `json:"userId"`
Modes string `json:"modes"`
JoinedAt time.Time `json:"joinedAt"`
Nick string `json:"nick"` // denormalized from users table
}
// User returns the full User for this membership.
func (cm *ChannelMember) User(ctx context.Context) (*User, error) {
if ul := cm.GetUserLookup(); ul != nil {
return ul.GetUserByID(ctx, cm.UserID)
}
return nil, fmt.Errorf("user lookup not available")
}
// Channel returns the full Channel for this membership.
func (cm *ChannelMember) Channel(ctx context.Context) (*Channel, error) {
if cl := cm.GetChannelLookup(); cl != nil {
return cl.GetChannelByID(ctx, cm.ChannelID)
}
return nil, fmt.Errorf("channel lookup not available")
}

20
internal/models/message.go
View File

@@ -1,20 +0,0 @@
package models
import (
"time"
)
// Message represents a chat message (channel or DM).
type Message struct {
Base
ID string `json:"id"`
Timestamp time.Time `json:"ts"`
FromUserID string `json:"fromUserId"`
FromNick string `json:"from"`
Target string `json:"to"`
Type string `json:"type"`
Body string `json:"body"`
Meta string `json:"meta"`
CreatedAt time.Time `json:"createdAt"`
}

15
internal/models/message_queue.go
View File

@@ -1,15 +0,0 @@
package models
import (
"time"
)
// MessageQueueEntry represents a pending message delivery for a user.
type MessageQueueEntry struct {
Base
ID int64 `json:"id"`
UserID string `json:"userId"`
MessageID string `json:"messageId"`
QueuedAt time.Time `json:"queuedAt"`
}

58
internal/models/model.go
View File

@@ -1,58 +0,0 @@
// Package models defines the data models used by the chat application.
// All model structs embed Base, which provides database access for
// relation-fetching methods directly on model instances.
package models
import (
"context"
"database/sql"
)
// DB is the interface that models use to query the database.
// This avoids a circular import with the db package.
type DB interface {
GetDB() *sql.DB
}
// UserLookup provides user lookup by ID without circular imports.
type UserLookup interface {
GetUserByID(ctx context.Context, id string) (*User, error)
}
// ChannelLookup provides channel lookup by ID without circular imports.
type ChannelLookup interface {
GetChannelByID(ctx context.Context, id string) (*Channel, error)
}
// Base is embedded in all model structs to provide database access.
type Base struct {
db DB
}
// SetDB injects the database reference into a model.
func (b *Base) SetDB(d DB) {
b.db = d
}
// GetDB returns the database interface for use in model methods.
func (b *Base) GetDB() *sql.DB {
return b.db.GetDB()
}
// GetUserLookup returns the DB as a UserLookup if it implements the interface.
func (b *Base) GetUserLookup() UserLookup {
if ul, ok := b.db.(UserLookup); ok {
return ul
}
return nil
}
// GetChannelLookup returns the DB as a ChannelLookup if it implements the interface.
func (b *Base) GetChannelLookup() ChannelLookup {
if cl, ok := b.db.(ChannelLookup); ok {
return cl
}
return nil
}

18
internal/models/server_link.go
View File

@@ -1,18 +0,0 @@
package models
import (
"time"
)
// ServerLink represents a federation peer server configuration.
type ServerLink struct {
Base
ID string `json:"id"`
Name string `json:"name"`
URL string `json:"url"`
SharedKeyHash string `json:"-"`
IsActive bool `json:"isActive"`
CreatedAt time.Time `json:"createdAt"`
LastSeenAt *time.Time `json:"lastSeenAt,omitempty"`
}

27
internal/models/session.go
View File

@@ -1,27 +0,0 @@
package models
import (
"context"
"fmt"
"time"
)
// Session represents a server-held user session.
type Session struct {
Base
ID string `json:"id"`
UserID string `json:"userId"`
CreatedAt time.Time `json:"createdAt"`
LastActiveAt time.Time `json:"lastActiveAt"`
ExpiresAt *time.Time `json:"expiresAt,omitempty"`
}
// User returns the user who owns this session.
func (s *Session) User(ctx context.Context) (*User, error) {
if ul := s.GetUserLookup(); ul != nil {
return ul.GetUserByID(ctx, s.UserID)
}
return nil, fmt.Errorf("user lookup not available")
}

92
internal/models/user.go
View File

@@ -1,92 +0,0 @@
package models
import (
"context"
"time"
)
// User represents a registered user account.
type User struct {
Base
ID string `json:"id"`
Nick string `json:"nick"`
PasswordHash string `json:"-"`
CreatedAt time.Time `json:"createdAt"`
UpdatedAt time.Time `json:"updatedAt"`
LastSeenAt *time.Time `json:"lastSeenAt,omitempty"`
}
// Channels returns all channels the user is a member of.
func (u *User) Channels(ctx context.Context) ([]*Channel, error) {
rows, err := u.GetDB().QueryContext(ctx, `
SELECT c.id, c.name, c.topic, c.modes, c.created_at, c.updated_at
FROM channels c
JOIN channel_members cm ON cm.channel_id = c.id
WHERE cm.user_id = ?
ORDER BY c.name`,
u.ID,
)
if err != nil {
return nil, err
}
defer func() { _ = rows.Close() }()
channels := []*Channel{}
for rows.Next() {
c := &Channel{}
c.SetDB(u.db)
err = rows.Scan(
&c.ID, &c.Name, &c.Topic, &c.Modes,
&c.CreatedAt, &c.UpdatedAt,
)
if err != nil {
return nil, err
}
channels = append(channels, c)
}
return channels, rows.Err()
}
// QueuedMessages returns undelivered messages for this user.
func (u *User) QueuedMessages(ctx context.Context) ([]*Message, error) {
rows, err := u.GetDB().QueryContext(ctx, `
SELECT m.id, m.ts, m.from_user_id, m.from_nick,
m.target, m.type, m.body, m.meta, m.created_at
FROM messages m
JOIN message_queue mq ON mq.message_id = m.id
WHERE mq.user_id = ?
ORDER BY mq.queued_at ASC`,
u.ID,
)
if err != nil {
return nil, err
}
defer func() { _ = rows.Close() }()
messages := []*Message{}
for rows.Next() {
msg := &Message{}
msg.SetDB(u.db)
err = rows.Scan(
&msg.ID, &msg.Timestamp, &msg.FromUserID,
&msg.FromNick, &msg.Target, &msg.Type,
&msg.Body, &msg.Meta, &msg.CreatedAt,
)
if err != nil {
return nil, err
}
messages = append(messages, msg)
}
return messages, rows.Err()
}

2
internal/server/http.go
View File

@@ -4,6 +4,6 @@ import "time"
const ( const (
httpReadTimeout = 10 * time.Second httpReadTimeout = 10 * time.Second
httpWriteTimeout = 10 * time.Second httpWriteTimeout = 60 * time.Second
maxHeaderBytes = 1 << 20 maxHeaderBytes = 1 << 20
) )

465
web/dist/app.js vendored
View File

File diff suppressed because one or more lines are too long