From eba6c2a02878acccbf84465257e5b2de43735953 Mon Sep 17 00:00:00 2001 From: dwrz Date: Sun, 20 Sep 2026 12:27:30 +0000 Subject: [PATCH] Add database package --- internal/db/db.go | 159 ++++++++ internal/db/db_test.go | 59 +++ internal/db/models.go | 82 +++++ internal/db/queries.go | 664 ++++++++++++++++++++++++++++++++++ internal/db/queries_test.go | 522 ++++++++++++++++++++++++++ internal/db/schema.go | 171 +++++++++ internal/db/schema/1.init.sql | 130 +++++++ internal/db/schema_test.go | 136 +++++++ 8 files changed, 1923 insertions(+) create mode 100644 internal/db/db.go create mode 100644 internal/db/db_test.go create mode 100644 internal/db/models.go create mode 100644 internal/db/queries.go create mode 100644 internal/db/queries_test.go create mode 100644 internal/db/schema.go create mode 100644 internal/db/schema/1.init.sql create mode 100644 internal/db/schema_test.go diff --git a/internal/db/db.go b/internal/db/db.go new file mode 100644 index 0000000..824ae05 --- /dev/null +++ b/internal/db/db.go @@ -0,0 +1,159 @@ +// Package db provides SQLite persistence for odidere. +package db + +import ( + "context" + "database/sql" + "fmt" + "regexp" + "sync" + + "github.com/mattn/go-sqlite3" +) + +// Config holds database connection settings. +type Config struct { + Path string +} + +// SQLite holds the reader and writer connection pools. +type SQLite struct { + Writer *sql.DB + Reader *sql.DB +} + +var driverOnce sync.Once + +// Connect opens a SQLite database with dual connection pools. +// +// Two connection pools are used: a single-connection writer and a +// multi-connection reader. This follows the recommended pattern for +// SQLite in WAL mode with Go's database/sql. +// +// The writer pool is limited to one connection. Go serializes +// access in-process, so SQLite never sees concurrent write +// attempts. This eliminates SQLITE_BUSY on writes without needing +// a busy_timeout. +// +// The reader pool allows concurrent reads, which WAL mode supports +// without blocking the writer or each other. +func Connect(ctx context.Context, cfg Config) (*SQLite, error) { + // Ensure timestamps default to RFC3339-compatible format. + sqlite3.SQLiteTimestampFormats = []string{ + "2006-01-02T15:04:05.000Z", + "2006-01-02T15:04:05.999999999-07:00", + "2006-01-02 15:04:05.999999999-07:00", + "2006-01-02 15:04:05.999999999", + "2006-01-02T15:04:05.999999999", + "2006-01-02 15:04:05", + "2006-01-02T15:04:05", + "2006-01-02 15:04", + "2006-01-02T15:04", + "2006-01-02", + } + + driverOnce.Do(func() { + sql.Register("odidere_sqlite", &sqlite3.SQLiteDriver{ + ConnectHook: func(conn *sqlite3.SQLiteConn) error { + if err := conn.RegisterFunc( + "regexp", + func(re, s string) (bool, error) { + return regexp.MatchString(re, s) + }, + true, + ); err != nil { + return fmt.Errorf("failed to register regexp: %w", err) + } + return nil + }, + }) + }) + + // Writer DSN. + // _busy_timeout=5000: wait up to 5 seconds if we encounter + // a lock. Rare in WAL mode, but possible during shutdown cleanup + // or crash recovery (see sqlite.org/wal.html §9). + // _cache_size=-16000: 16 MB page cache. Reduces disk reads for + // the single write connection that handles all mutations. + // _foreign_keys=on: enforce foreign key constraints. + // _journal_mode=WAL: write-ahead logging. Allows concurrent + // readers while writing. + // _synchronous=1: NORMAL durability. WAL mode ensures consistency + // even with NORMAL. Avoids the performance cost of FULL + // (fsync every commit). + // _txlock=immediate: acquire the write lock at BEGIN, not at the + // first write statement. Prevents deadlocks when multiple + // connections attempt deferred → write lock escalation. + writerDSN := "file:" + cfg.Path + "?" + + "_busy_timeout=5000&" + + "_cache_size=-16000&" + + "_foreign_keys=on&" + + "_journal_mode=WAL&" + + "_loc=UTC&" + + "_synchronous=1&" + + "_txlock=immediate" + w, err := sql.Open("odidere_sqlite", writerDSN) + if err != nil { + return nil, fmt.Errorf( + "failed to open sqlite writer: %v", err, + ) + } + // One connection: all writes serialize in-process via the pool. + w.SetMaxOpenConns(1) + w.SetMaxIdleConns(1) + if err := w.PingContext(ctx); err != nil { + return nil, fmt.Errorf( + "failed to ping sqlite writer: %v", err, + ) + } + + // Reader DSN. + // mode=ro: open read-only. Prevents accidental writes through + // the reader pool. + // _busy_timeout=5000: wait up to 5 seconds if a read encounters + // a lock. Rare in WAL mode, but possible during shutdown cleanup + // or crash recovery (see sqlite.org/wal.html §9). + // _cache_size=-8000: 8 MB page cache per connection. Smaller than + // the writer since there are multiple reader connections. + // _journal_mode=WAL: must match the writer. WAL is persistent on + // the file, but setting it here ensures the connection is + // configured correctly if the database is new. + readerDSN := "file:" + cfg.Path + "?" + + "_busy_timeout=5000&" + + "_cache_size=-8000&" + + "_journal_mode=WAL&" + + "_loc=UTC&" + + "mode=ro" + r, err := sql.Open("odidere_sqlite", readerDSN) + if err != nil { + return nil, fmt.Errorf( + "failed to open sqlite reader: %v", err, + ) + } + r.SetMaxOpenConns(2) + r.SetMaxIdleConns(2) + if err := r.PingContext(ctx); err != nil { + return nil, fmt.Errorf( + "failed to ping sqlite reader: %v", err, + ) + } + + // Keep the connections open. + w.SetConnMaxLifetime(0) + w.SetConnMaxIdleTime(0) + r.SetConnMaxLifetime(0) + r.SetConnMaxIdleTime(0) + + return &SQLite{Reader: r, Writer: w}, nil +} + +// Close shuts down both connection pools. +func (s *SQLite) Close() error { + if err := s.Writer.Close(); err != nil { + return fmt.Errorf("failed to close writer: %w", err) + } + if err := s.Reader.Close(); err != nil { + return fmt.Errorf("failed to close reader: %w", err) + } + return nil +} diff --git a/internal/db/db_test.go b/internal/db/db_test.go new file mode 100644 index 0000000..8aa6703 --- /dev/null +++ b/internal/db/db_test.go @@ -0,0 +1,59 @@ +package db + +import ( + "context" + "os" + "path/filepath" + "testing" +) + +// createDB sets up a real SQLite database in a temporary directory +// and returns the connection. The temporary directory is cleaned up +// when the test ends. The caller is responsible for applying the schema. +func createDB(t *testing.T) *SQLite { + t.Helper() + + var ( + dir = t.TempDir() + path = filepath.Join(dir, "test.db") + ) + db, err := Connect(t.Context(), Config{Path: path}) + if err != nil { + t.Fatalf("Connect: %v", err) + } + + return db +} + +func TestDB(t *testing.T) { + testConnect(t, t.Context()) +} + +func testConnect(t *testing.T, ctx context.Context) { + // Test bad path. + _, err := Connect(ctx, Config{Path: "/nonexistent/directory/test.db"}) + if err == nil { + t.Fatal("expected error connecting to bad path") + } + + // Test connect. + path := filepath.Join(t.TempDir(), "test.db") + db, err := Connect(ctx, Config{Path: path}) + if err != nil { + t.Fatalf("Connect: %v", err) + } + if db.Writer == nil { + t.Fatal("Writer is nil") + } + if db.Reader == nil { + t.Fatal("Reader is nil") + } + + // Test close. + if err := db.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + if _, err := os.Stat(path); os.IsNotExist(err) { + t.Fatal("database file does not exist") + } +} diff --git a/internal/db/models.go b/internal/db/models.go new file mode 100644 index 0000000..a94daff --- /dev/null +++ b/internal/db/models.go @@ -0,0 +1,82 @@ +package db + +import "time" + +// ApprovalStatus defines the lifecycle states for a tool call approval. +type ApprovalStatus string + +const ( + ApprovalStatusApproved ApprovalStatus = "approved" + ApprovalStatusDenied ApprovalStatus = "denied" + ApprovalStatusPending ApprovalStatus = "pending" +) + +// Approval represents a tool call awaiting or resolved by user permission. +type Approval struct { + CreatedAt time.Time `json:"created_at"` + Id string `json:"id"` + MessageId string `json:"message_id"` + Status ApprovalStatus `json:"status"` + ToolCallId string `json:"tool_call_id"` + UpdatedAt time.Time `json:"updated_at"` +} + +// Conversation represents a stored conversation. +type Conversation struct { + CreatedAt time.Time `json:"created_at"` + Id string `json:"id"` + Title *string `json:"title"` + UpdatedAt time.Time `json:"updated_at"` + UserId string `json:"user_id"` +} + +// MessageRole defines the permitted roles for messages. +type MessageRole string + +const ( + MessageRoleSystem MessageRole = "system" + MessageRoleDeveloper MessageRole = "developer" + MessageRoleUser MessageRole = "user" + MessageRoleAssistant MessageRole = "assistant" + MessageRoleTool MessageRole = "tool" +) + +// Message represents a stored chat message. Content holds the +// wire-format JSON for the message — the exact JSON object that +// appears in the OpenAI Chat Completions messages array. +type Message struct { + Content string `json:"content"` + ConversationId string `json:"conversation_id"` + CreatedAt time.Time `json:"created_at"` + Id string `json:"id"` + Role MessageRole `json:"role"` + Sequence int `json:"sequence"` + UpdatedAt time.Time `json:"updated_at"` +} + +// Session represents a user login session. +type Session struct { + CreatedAt time.Time `json:"created_at"` + DeactivatedAt *time.Time `json:"deactivated_at"` + Id string `json:"id"` + UpdatedAt time.Time `json:"updated_at"` + UserId string `json:"user_id"` +} + +// UserRole defines the permitted roles for users. +type UserRole string + +const ( + UserRoleOwner UserRole = "owner" +) + +// User represents a stored user account. +type User struct { + CreatedAt time.Time `json:"created_at"` + DeactivatedAt *time.Time `json:"deactivated_at"` + Id string `json:"id"` + Name string `json:"name"` + Password string `json:"password"` + Role UserRole `json:"role"` + UpdatedAt time.Time `json:"updated_at"` +} diff --git a/internal/db/queries.go b/internal/db/queries.go new file mode 100644 index 0000000..37d3b03 --- /dev/null +++ b/internal/db/queries.go @@ -0,0 +1,664 @@ +package db + +import ( + "context" + "database/sql" + "fmt" + "time" + "uuid" +) + +// DBTX is the interface satisfied by both *sql.DB and *sql.Tx. +type DBTX interface { + ExecContext(context.Context, string, ...any) (sql.Result, error) + QueryContext(context.Context, string, ...any) (*sql.Rows, error) + QueryRowContext(context.Context, string, ...any) *sql.Row +} + +// Queries provides database query methods. +type Queries struct{ db DBTX } + +// New creates a Queries backed by the given DBTX. +func New(db DBTX) *Queries { return &Queries{db: db} } + +// WithTx returns a Queries that runs within the given transaction. +func (q *Queries) WithTx(tx *sql.Tx) *Queries { return &Queries{db: tx} } + +const insertUser = ` +INSERT INTO users (id, name, password, role) +VALUES (?, ?, ?, ?) +RETURNING created_at, deactivated_at, id, name, password, role, updated_at +` + +// InsertUserParams holds arguments for InsertUser. +type InsertUserParams struct { + Name string + Password string + Role UserRole +} + +// InsertUser creates a new user. +func (q *Queries) InsertUser(ctx context.Context, arg InsertUserParams) (*User, error) { + row := q.db.QueryRowContext( + ctx, + insertUser, + uuid.New().String(), + arg.Name, + arg.Password, + arg.Role, + ) + + var ( + u User + createdAt string + updatedAt string + err error + ) + if err := row.Scan( + &createdAt, + &u.DeactivatedAt, + &u.Id, + &u.Name, + &u.Password, + &u.Role, + &updatedAt, + ); err != nil { + return nil, fmt.Errorf( + "InsertUser: %w", err, + ) + } + if u.CreatedAt, err = time.Parse(time.RFC3339, createdAt); err != nil { + return nil, fmt.Errorf("InsertUser: parse created_at: %w", err) + } + if u.UpdatedAt, err = time.Parse(time.RFC3339, updatedAt); err != nil { + return nil, fmt.Errorf("InsertUser: parse updated_at: %w", err) + } + return &u, nil +} + +const selectUserId = ` +SELECT created_at, deactivated_at, id, name, password, role, updated_at +FROM users +WHERE id = ? +` + +// SelectUserId returns a single user by ID. +func (q *Queries) SelectUserId(ctx context.Context, id string) (*User, error) { + row := q.db.QueryRowContext(ctx, selectUserId, id) + + var ( + u User + createdAt string + updatedAt string + ) + err := row.Scan( + &createdAt, + &u.DeactivatedAt, + &u.Id, + &u.Name, + &u.Password, + &u.Role, + &updatedAt, + ) + if err != nil { + return nil, fmt.Errorf("SelectUserId: %w", err) + } + + if u.CreatedAt, err = time.Parse(time.RFC3339, createdAt); err != nil { + return nil, fmt.Errorf("SelectUserId: parse created_at: %w", err) + } + if u.UpdatedAt, err = time.Parse(time.RFC3339, updatedAt); err != nil { + return nil, fmt.Errorf("SelectUserId: parse updated_at: %w", err) + } + + return &u, nil +} + +const insertConversation = ` +INSERT INTO conversations (id, title, user_id) +VALUES (?, ?, ?) +RETURNING created_at, id, title, updated_at, user_id +` + +// InsertConversationParams holds arguments for InsertConversation. +type InsertConversationParams struct { + Title *string + UserId string +} + +// InsertConversation creates a new conversation. +func (q *Queries) InsertConversation(ctx context.Context, arg InsertConversationParams) (*Conversation, error) { + row := q.db.QueryRowContext( + ctx, + insertConversation, + uuid.New().String(), + arg.Title, + arg.UserId, + ) + + var ( + c Conversation + createdAt string + updatedAt string + err error + ) + if err := row.Scan( + &createdAt, + &c.Id, + &c.Title, + &updatedAt, + &c.UserId, + ); err != nil { + return nil, fmt.Errorf("InsertConversation: %w", err) + } + if c.CreatedAt, err = time.Parse(time.RFC3339, createdAt); err != nil { + return nil, fmt.Errorf("InsertConversation: parse created_at: %w", err) + } + if c.UpdatedAt, err = time.Parse(time.RFC3339, updatedAt); err != nil { + return nil, fmt.Errorf("InsertConversation: parse updated_at: %w", err) + } + return &c, nil +} + +const selectConversationId = ` +SELECT created_at, id, title, updated_at, user_id +FROM conversations +WHERE id = ? +` + +// SelectConversationId returns a single conversation by ID. +func (q *Queries) SelectConversationId(ctx context.Context, id string) (*Conversation, error) { + row := q.db.QueryRowContext(ctx, selectConversationId, id) + + var ( + c Conversation + createdAt string + updatedAt string + ) + err := row.Scan( + &createdAt, + &c.Id, + &c.Title, + &updatedAt, + &c.UserId, + ) + if err != nil { + return nil, fmt.Errorf("SelectConversationId: %w", err) + } + if c.CreatedAt, err = time.Parse(time.RFC3339, createdAt); err != nil { + return nil, fmt.Errorf("SelectConversationId: parse created_at: %w", err) + } + if c.UpdatedAt, err = time.Parse(time.RFC3339, updatedAt); err != nil { + return nil, fmt.Errorf("SelectConversationId: parse updated_at: %w", err) + } + + return &c, nil +} + +const selectConversationIdUserId = ` +SELECT created_at, id, title, updated_at, user_id +FROM conversations +WHERE id = ? AND user_id = ? +` + +// SelectConversationIdUserId returns a conversation by ID when it belongs to +// the specified user. +func (q *Queries) SelectConversationIdUserId( + ctx context.Context, + id string, + userId string, +) (*Conversation, error) { + row := q.db.QueryRowContext(ctx, selectConversationIdUserId, id, userId) + + var ( + c Conversation + createdAt string + updatedAt string + ) + err := row.Scan( + &createdAt, + &c.Id, + &c.Title, + &updatedAt, + &c.UserId, + ) + if err != nil { + return nil, fmt.Errorf("SelectConversationIdUserId: %w", err) + } + if c.CreatedAt, err = time.Parse(time.RFC3339, createdAt); err != nil { + return nil, fmt.Errorf( + "SelectConversationIdUserId: parse created_at: %w", err, + ) + } + if c.UpdatedAt, err = time.Parse(time.RFC3339, updatedAt); err != nil { + return nil, fmt.Errorf( + "SelectConversationIdUserId: parse updated_at: %w", err, + ) + } + + return &c, nil +} + +const selectConversationsUserId = ` +SELECT created_at, id, title, updated_at, user_id +FROM conversations +WHERE user_id = ? +ORDER BY updated_at DESC +` + +// SelectConversationsUserId returns all conversations for a user, +// ordered by most recently updated. +func (q *Queries) SelectConversationsUserId(ctx context.Context, userId string) ([]Conversation, error) { + rows, err := q.db.QueryContext(ctx, selectConversationsUserId, userId) + if err != nil { + return nil, fmt.Errorf("SelectConversationsUserId: %w", err) + } + defer rows.Close() + + var conversations []Conversation + for rows.Next() { + var ( + c Conversation + createdAt string + updatedAt string + ) + if err := rows.Scan( + &createdAt, + &c.Id, + &c.Title, + &updatedAt, + &c.UserId, + ); err != nil { + return nil, fmt.Errorf("SelectConversationsUserId: %w", err) + } + if c.CreatedAt, err = time.Parse(time.RFC3339, createdAt); err != nil { + return nil, fmt.Errorf("SelectConversationsUserId: parse created_at: %w", err) + } + if c.UpdatedAt, err = time.Parse(time.RFC3339, updatedAt); err != nil { + return nil, fmt.Errorf("SelectConversationsUserId: parse updated_at: %w", err) + } + conversations = append(conversations, c) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("SelectConversationsUserId: %w", err) + } + return conversations, nil +} + +const deleteConversationId = ` +DELETE FROM conversations +WHERE id = ? +` + +// DeleteConversation deletes a conversation by ID. Cascades to +// all messages in the conversation. +func (q *Queries) DeleteConversationId(ctx context.Context, id string) error { + _, err := q.db.ExecContext(ctx, deleteConversationId, id) + if err != nil { + return fmt.Errorf("DeleteConversationId: %w", err) + } + return nil +} + +const insertMessage = ` +INSERT INTO messages (id, content, conversation_id, role, sequence) +VALUES (?, ?, ?, ?, ?) +RETURNING content, conversation_id, created_at, id, role, sequence, updated_at +` + +// InsertMessageParams holds arguments for InsertMessage. +type InsertMessageParams struct { + Content string + ConversationId string + Role MessageRole + Sequence int +} + +// InsertMessage creates a new message. +func (q *Queries) InsertMessage(ctx context.Context, arg InsertMessageParams) (*Message, error) { + row := q.db.QueryRowContext( + ctx, + insertMessage, + uuid.New().String(), + arg.Content, + arg.ConversationId, + arg.Role, + arg.Sequence, + ) + + var ( + m Message + createdAt string + updatedAt string + err error + ) + if err := row.Scan( + &m.Content, + &m.ConversationId, + &createdAt, + &m.Id, + &m.Role, + &m.Sequence, + &updatedAt, + ); err != nil { + return nil, fmt.Errorf("InsertMessage: %w", err) + } + if m.CreatedAt, err = time.Parse(time.RFC3339, createdAt); err != nil { + return nil, fmt.Errorf("InsertMessage: parse created_at: %w", err) + } + if m.UpdatedAt, err = time.Parse(time.RFC3339, updatedAt); err != nil { + return nil, fmt.Errorf("InsertMessage: parse updated_at: %w", err) + } + return &m, nil +} + +const selectMessagesConversationId = ` +SELECT content, conversation_id, created_at, id, role, sequence, updated_at +FROM messages +WHERE conversation_id = ? +ORDER BY sequence +` + +// SelectMessagesConversationId returns all messages for a +// conversation, ordered by sequence. +func (q *Queries) SelectMessagesConversationId(ctx context.Context, conversationId string) ([]Message, error) { + rows, err := q.db.QueryContext(ctx, selectMessagesConversationId, conversationId) + if err != nil { + return nil, fmt.Errorf("SelectMessagesConversationId: %w", err) + } + defer rows.Close() + + var messages []Message + for rows.Next() { + var ( + m Message + createdAt string + updatedAt string + ) + if err := rows.Scan( + &m.Content, + &m.ConversationId, + &createdAt, + &m.Id, + &m.Role, + &m.Sequence, + &updatedAt, + ); err != nil { + return nil, fmt.Errorf("SelectMessagesConversationId: %w", err) + } + if m.CreatedAt, err = time.Parse(time.RFC3339, createdAt); err != nil { + return nil, fmt.Errorf("SelectMessagesConversationId: parse created_at: %w", err) + } + if m.UpdatedAt, err = time.Parse(time.RFC3339, updatedAt); err != nil { + return nil, fmt.Errorf("SelectMessagesConversationId: parse updated_at: %w", err) + } + messages = append(messages, m) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("SelectMessagesConversationId: %w", err) + } + return messages, nil +} + +const updateMessageContent = ` +UPDATE messages +SET content = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') +WHERE id = ? +` + +// UpdateMessageContentParams holds arguments for UpdateMessageContent. +type UpdateMessageContentParams struct { + Content string + Id string +} + +// UpdateMessageContent sets the content JSON for a message. +func (q *Queries) UpdateMessageContent(ctx context.Context, arg UpdateMessageContentParams) error { + _, err := q.db.ExecContext( + ctx, + updateMessageContent, + arg.Content, + arg.Id, + ) + if err != nil { + return fmt.Errorf("UpdateMessageContent: %w", err) + } + return nil +} + +const deleteMessageId = ` +DELETE FROM messages +WHERE id = ? +` + +// DeleteMessage deletes a message by ID. +func (q *Queries) DeleteMessageId(ctx context.Context, id string) error { + _, err := q.db.ExecContext(ctx, deleteMessageId, id) + if err != nil { + return fmt.Errorf("DeleteMessageId: %w", err) + } + return nil +} + +const selectMaxSequence = ` +SELECT COALESCE(MAX(sequence), 0) +FROM messages +WHERE conversation_id = ? +` + +// NextSequence returns the next available sequence number for +// a conversation. +func (q *Queries) NextSequence(ctx context.Context, conversationId string) (int, error) { + var seq int + err := q.db.QueryRowContext(ctx, selectMaxSequence, conversationId).Scan(&seq) + if err != nil { + return 0, fmt.Errorf("NextSequence: %w", err) + } + return seq + 1, nil +} + +const insertApproval = ` +INSERT INTO approvals (id, message_id, tool_call_id) +VALUES (?, ?, ?) +RETURNING created_at, id, message_id, status, tool_call_id, updated_at +` + +// InsertApprovalParams holds arguments for InsertApproval. +type InsertApprovalParams struct { + MessageId string + ToolCallId string +} + +// InsertApproval creates a new pending approval. +func (q *Queries) InsertApproval(ctx context.Context, arg InsertApprovalParams) (*Approval, error) { + row := q.db.QueryRowContext( + ctx, + insertApproval, + uuid.New().String(), + arg.MessageId, + arg.ToolCallId, + ) + + var ( + a Approval + createdAt string + updatedAt string + err error + ) + if err = row.Scan( + &createdAt, + &a.Id, + &a.MessageId, + &a.Status, + &a.ToolCallId, + &updatedAt, + ); err != nil { + return nil, fmt.Errorf( + "InsertApproval: %w", err, + ) + } + if a.CreatedAt, err = time.Parse(time.RFC3339, createdAt); err != nil { + return nil, fmt.Errorf("InsertApproval: parse created_at: %w", err) + } + if a.UpdatedAt, err = time.Parse(time.RFC3339, updatedAt); err != nil { + return nil, fmt.Errorf("InsertApproval: parse updated_at: %w", err) + } + return &a, nil +} + +const selectApprovalId = ` +SELECT created_at, id, message_id, status, tool_call_id, updated_at +FROM approvals +WHERE id = ? +` + +// SelectApprovalId returns a single approval by ID. +func (q *Queries) SelectApprovalId(ctx context.Context, id string) (*Approval, error) { + row := q.db.QueryRowContext(ctx, selectApprovalId, id) + + var ( + a Approval + createdAt string + updatedAt string + ) + err := row.Scan( + &createdAt, + &a.Id, + &a.MessageId, + &a.Status, + &a.ToolCallId, + &updatedAt, + ) + if err != nil { + return nil, fmt.Errorf("SelectApprovalId: %w", err) + } + if a.CreatedAt, err = time.Parse(time.RFC3339, createdAt); err != nil { + return nil, fmt.Errorf("SelectApprovalId: parse created_at: %w", err) + } + if a.UpdatedAt, err = time.Parse(time.RFC3339, updatedAt); err != nil { + return nil, fmt.Errorf("SelectApprovalId: parse updated_at: %w", err) + } + return &a, nil +} + +const selectApprovalsMessageId = ` +SELECT created_at, id, message_id, status, tool_call_id, updated_at +FROM approvals +WHERE message_id = ? +ORDER BY created_at +` + +// SelectApprovalsMessageId returns all approvals for a message, +// ordered by creation time. +func (q *Queries) SelectApprovalsMessageId(ctx context.Context, messageId string) ([]Approval, error) { + rows, err := q.db.QueryContext(ctx, selectApprovalsMessageId, messageId) + if err != nil { + return nil, fmt.Errorf("SelectApprovalsMessageId: %w", err) + } + defer rows.Close() + + var approvals []Approval + for rows.Next() { + var ( + a Approval + createdAt string + updatedAt string + ) + if err := rows.Scan( + &createdAt, + &a.Id, + &a.MessageId, + &a.Status, + &a.ToolCallId, + &updatedAt, + ); err != nil { + return nil, fmt.Errorf("SelectApprovalsMessageId: %w", err) + } + if a.CreatedAt, err = time.Parse(time.RFC3339, createdAt); err != nil { + return nil, fmt.Errorf("SelectApprovalsMessageId: parse created_at: %w", err) + } + if a.UpdatedAt, err = time.Parse(time.RFC3339, updatedAt); err != nil { + return nil, fmt.Errorf("SelectApprovalsMessageId: parse updated_at: %w", err) + } + approvals = append(approvals, a) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("SelectApprovalsMessageId: %w", err) + } + return approvals, nil +} + +const selectPendingApprovals = ` +SELECT created_at, id, message_id, status, tool_call_id, updated_at +FROM approvals +WHERE status = 'pending' +ORDER BY created_at +` + +// SelectPendingApprovals returns all pending approvals across all +// conversations, ordered by creation time. This serves as the global +// inbox for the UI. +func (q *Queries) SelectPendingApprovals(ctx context.Context) ([]Approval, error) { + rows, err := q.db.QueryContext(ctx, selectPendingApprovals) + if err != nil { + return nil, fmt.Errorf("SelectPendingApprovals: %w", err) + } + defer rows.Close() + + var approvals []Approval + for rows.Next() { + var ( + a Approval + createdAt string + updatedAt string + ) + if err := rows.Scan( + &createdAt, + &a.Id, + &a.MessageId, + &a.Status, + &a.ToolCallId, + &updatedAt, + ); err != nil { + return nil, fmt.Errorf("SelectPendingApprovals: %w", err) + } + if a.CreatedAt, err = time.Parse(time.RFC3339, createdAt); err != nil { + return nil, fmt.Errorf("SelectPendingApprovals: parse created_at: %w", err) + } + if a.UpdatedAt, err = time.Parse(time.RFC3339, updatedAt); err != nil { + return nil, fmt.Errorf("SelectPendingApprovals: parse updated_at: %w", err) + } + approvals = append(approvals, a) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("SelectPendingApprovals: %w", err) + } + return approvals, nil +} + +const updateApprovalStatus = ` +UPDATE approvals +SET status = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') +WHERE id = ? +` + +// UpdateApprovalStatusParams holds arguments for UpdateApprovalStatus. +type UpdateApprovalStatusParams struct { + Id string + Status ApprovalStatus +} + +// UpdateApprovalStatus transitions an approval's status and bumps +// updated_at. The caller is responsible for valid transitions. +func (q *Queries) UpdateApprovalStatus(ctx context.Context, arg UpdateApprovalStatusParams) error { + _, err := q.db.ExecContext( + ctx, + updateApprovalStatus, + arg.Status, + arg.Id, + ) + if err != nil { + return fmt.Errorf("UpdateApprovalStatus: %w", err) + } + return nil +} diff --git a/internal/db/queries_test.go b/internal/db/queries_test.go new file mode 100644 index 0000000..7665678 --- /dev/null +++ b/internal/db/queries_test.go @@ -0,0 +1,522 @@ +package db + +import ( + "context" + "testing" +) + +func TestQueries(t *testing.T) { + var ( + ctx = t.Context() + db = createDB(t) + ) + defer db.Close() + + if err := db.ApplySchema(ctx, nil); err != nil { + t.Fatalf("ApplySchema: %v", err) + } + + t.Run("User", func(t *testing.T) { + testUser(t, ctx, db) + }) + t.Run("Conversation", func(t *testing.T) { + testConversation(t, ctx, db) + }) + t.Run("Messaging", func(t *testing.T) { + testMessaging(t, ctx, db) + }) + t.Run("ConversationTimestamp", func(t *testing.T) { + testConversationTimestamp(t, ctx, db) + }) + t.Run("Approvals", func(t *testing.T) { + testApprovals(t, ctx, db) + }) + t.Run("Delete", func(t *testing.T) { + testDelete(t, ctx, db) + }) +} + +func testUser(t *testing.T, ctx context.Context, db *SQLite) { + var ( + qr = New(db.Reader) + qw = New(db.Writer) + ) + u, err := qw.InsertUser(ctx, InsertUserParams{ + Name: "user", + Password: "test", + Role: UserRoleOwner, + }) + if err != nil { + t.Fatalf("InsertUser: %v", err) + } + + res, err := qr.SelectUserId(ctx, u.Id) + if err != nil { + t.Fatalf("SelectUserId: %v", err) + } + if res == nil { + t.Fatal("SelectUserId returned nil") + } + if res.Name != u.Name { + t.Errorf( + "expected name %s, got %s", + u.Name, res.Name, + ) + } +} + +func testConversation(t *testing.T, ctx context.Context, db *SQLite) { + var ( + qr = New(db.Reader) + qw = New(db.Writer) + title = "test" + ) + + u, err := qw.InsertUser(ctx, InsertUserParams{ + Name: "conversation", + Password: "test", + Role: UserRoleOwner, + }) + if err != nil { + t.Fatalf("InsertUser: %v", err) + } + + c, err := qw.InsertConversation(ctx, InsertConversationParams{ + UserId: u.Id, + Title: &title, + }) + if err != nil { + t.Fatalf("InsertConversation: %v", err) + } + + res, err := qr.SelectConversationId(ctx, c.Id) + if err != nil { + t.Fatalf("SelectConversationId: %v", err) + } + if res == nil { + t.Fatal("SelectConversationId returned nil") + } + if res.Title == nil || *res.Title != title { + t.Errorf("expected title %s, got %v", title, res.Title) + } + + owned, err := qr.SelectConversationIdUserId(ctx, c.Id, u.Id) + if err != nil { + t.Fatalf("SelectConversationIdUserId: %v", err) + } + if owned.Id != c.Id { + t.Errorf("expected conversation %s, got %s", c.Id, owned.Id) + } + if _, err := qr.SelectConversationIdUserId(ctx, c.Id, "unknown"); err == nil { + t.Fatal("expected error selecting conversation for wrong user") + } + if _, err := qr.SelectConversationIdUserId(ctx, "unknown", u.Id); err == nil { + t.Fatal("expected error selecting unknown conversation") + } + + // Test conversation list. + conversations, err := qr.SelectConversationsUserId(ctx, u.Id) + if err != nil { + t.Fatalf("SelectConversationsUserId: %v", err) + } + if len(conversations) != 1 { + t.Fatalf("expected 1 conversation, got %d", len(conversations)) + } +} + +func testMessaging(t *testing.T, ctx context.Context, db *SQLite) { + var qw = New(db.Writer) + + u, err := qw.InsertUser(ctx, InsertUserParams{ + Name: "messaging", + Password: "test", + Role: UserRoleOwner, + }) + if err != nil { + t.Fatalf("InsertUser: %v", err) + } + + c, err := qw.InsertConversation(ctx, InsertConversationParams{ + UserId: u.Id, + }) + if err != nil { + t.Fatalf("InsertConversation: %v", err) + } + + // Insert a system message. + seq, err := qw.NextSequence(ctx, c.Id) + if err != nil { + t.Fatalf("NextSequence: %v", err) + } + _, err = qw.InsertMessage(ctx, InsertMessageParams{ + ConversationId: c.Id, + Sequence: seq, + Role: MessageRoleSystem, + Content: `{"role":"system","content":"You are helpful"}`, + }) + if err != nil { + t.Fatalf("InsertMessage: %v", err) + } + + // Insert a user message. + seq, err = qw.NextSequence(ctx, c.Id) + if err != nil { + t.Fatalf("NextSequence: %v", err) + } + _, err = qw.InsertMessage(ctx, InsertMessageParams{ + ConversationId: c.Id, + Sequence: seq, + Role: MessageRoleUser, + Content: `{"role":"user","content":"Hello"}`, + }) + if err != nil { + t.Fatalf("InsertMessage: %v", err) + } + + // Insert an assistant message with tool calls. + seq, err = qw.NextSequence(ctx, c.Id) + if err != nil { + t.Fatalf("NextSequence: %v", err) + } + _, err = qw.InsertMessage(ctx, InsertMessageParams{ + ConversationId: c.Id, + Sequence: seq, + Role: MessageRoleAssistant, + Content: `{"role":"assistant","content":null,"tool_calls":[{"id":"call_abc","type":"function","function":{"name":"get_weather","arguments":"{\"location\":\"NYC\"}"}}]}`, + }) + if err != nil { + t.Fatalf("InsertMessage: %v", err) + } + + // Insert a tool result message. + seq, err = qw.NextSequence(ctx, c.Id) + if err != nil { + t.Fatalf("NextSequence: %v", err) + } + toolMsg, err := qw.InsertMessage(ctx, InsertMessageParams{ + ConversationId: c.Id, + Sequence: seq, + Role: MessageRoleTool, + Content: `{"role":"tool","content":"{\"temperature\":72}","tool_call_id":"call_abc"}`, + }) + if err != nil { + t.Fatalf("InsertMessage: %v", err) + } + + // Test SelectMessages. + qr := New(db.Reader) + messages, err := qr.SelectMessagesConversationId(ctx, c.Id) + if err != nil { + t.Fatalf("SelectMessagesConversationId: %v", err) + } + if len(messages) != 4 { + t.Fatalf("expected 4 messages, got %d", len(messages)) + } + if messages[0].Role != MessageRoleSystem { + t.Errorf("expected first message role system, got %s", messages[0].Role) + } + if messages[3].Role != MessageRoleTool { + t.Errorf("expected last message role tool, got %s", messages[3].Role) + } + + // Test UpdateMessageContent. + if err := qw.UpdateMessageContent(ctx, UpdateMessageContentParams{ + Id: toolMsg.Id, + Content: `{"role":"tool","content":"{\"temperature\":75}","tool_call_id":"call_abc"}`, + }); err != nil { + t.Fatalf("UpdateMessageContent: %v", err) + } + + updated, err := qr.SelectMessagesConversationId(ctx, c.Id) + if err != nil { + t.Fatalf("SelectMessagesConversationId: %v", err) + } + if updated[3].Content != `{"role":"tool","content":"{\"temperature\":75}","tool_call_id":"call_abc"}` { + t.Errorf("updated content mismatch: %s", updated[3].Content) + } +} + +func testConversationTimestamp(t *testing.T, ctx context.Context, database *SQLite) { + var ( + qr = New(database.Reader) + qw = New(database.Writer) + ) + + u, err := qw.InsertUser(ctx, InsertUserParams{ + Name: "conversation-timestamp", + Password: "test", + Role: UserRoleOwner, + }) + if err != nil { + t.Fatalf("InsertUser: %v", err) + } + c, err := qw.InsertConversation(ctx, InsertConversationParams{UserId: u.Id}) + if err != nil { + t.Fatalf("InsertConversation: %v", err) + } + + const oldTimestamp = "2000-01-01T00:00:00.000Z" + if _, err := database.Writer.ExecContext( + ctx, + "UPDATE conversations SET updated_at = ? WHERE id = ?", + oldTimestamp, + c.Id, + ); err != nil { + t.Fatalf("set conversation timestamp before insert: %v", err) + } + msg, err := qw.InsertMessage(ctx, InsertMessageParams{ + ConversationId: c.Id, + Sequence: 0, + Role: MessageRoleUser, + Content: `{"role":"user","content":"Hello"}`, + }) + if err != nil { + t.Fatalf("InsertMessage: %v", err) + } + conv, err := qr.SelectConversationId(ctx, c.Id) + if err != nil { + t.Fatalf("SelectConversationId after message insert: %v", err) + } + if conv.UpdatedAt.Format("2006-01-02T15:04:05.000Z") == oldTimestamp { + t.Error("message insert did not update conversation timestamp") + } + + if _, err := database.Writer.ExecContext( + ctx, + "UPDATE conversations SET updated_at = ? WHERE id = ?", + oldTimestamp, + c.Id, + ); err != nil { + t.Fatalf("set conversation timestamp before update: %v", err) + } + if err := qw.UpdateMessageContent(ctx, UpdateMessageContentParams{ + Id: msg.Id, + Content: `{"role":"user","content":"Updated"}`, + }); err != nil { + t.Fatalf("UpdateMessageContent: %v", err) + } + conv, err = qr.SelectConversationId(ctx, c.Id) + if err != nil { + t.Fatalf("SelectConversationId after message update: %v", err) + } + if conv.UpdatedAt.Format("2006-01-02T15:04:05.000Z") == oldTimestamp { + t.Error("message update did not update conversation timestamp") + } + + if _, err := database.Writer.ExecContext( + ctx, + "UPDATE conversations SET updated_at = ? WHERE id = ?", + oldTimestamp, + c.Id, + ); err != nil { + t.Fatalf("set conversation timestamp before delete: %v", err) + } + if err := qw.DeleteMessageId(ctx, msg.Id); err != nil { + t.Fatalf("DeleteMessageId: %v", err) + } + conv, err = qr.SelectConversationId(ctx, c.Id) + if err != nil { + t.Fatalf("SelectConversationId after message delete: %v", err) + } + if conv.UpdatedAt.Format("2006-01-02T15:04:05.000Z") == oldTimestamp { + t.Error("message delete did not update conversation timestamp") + } +} + +func testDelete(t *testing.T, ctx context.Context, db *SQLite) { + var ( + qr = New(db.Reader) + qw = New(db.Writer) + ) + + u, err := qw.InsertUser(ctx, InsertUserParams{ + Name: "delete", + Password: "test", + Role: UserRoleOwner, + }) + if err != nil { + t.Fatalf("InsertUser: %v", err) + } + + c, err := qw.InsertConversation(ctx, InsertConversationParams{ + UserId: u.Id, + }) + if err != nil { + t.Fatalf("InsertConversation: %v", err) + } + + msg, err := qw.InsertMessage(ctx, InsertMessageParams{ + ConversationId: c.Id, + Sequence: 1, + Role: MessageRoleUser, + Content: `{"role":"user","content":"Hello"}`, + }) + if err != nil { + t.Fatalf("InsertMessage: %v", err) + } + + // Delete message. + if err := qw.DeleteMessageId(ctx, msg.Id); err != nil { + t.Fatalf("DeleteMessageId: %v", err) + } + + messages, err := qr.SelectMessagesConversationId(ctx, c.Id) + if err != nil { + t.Fatalf("SelectMessagesConversationId: %v", err) + } + if len(messages) != 0 { + t.Fatalf("expected 0 messages after delete, got %d", len(messages)) + } + + // Delete conversation. + if err := qw.DeleteConversationId(ctx, c.Id); err != nil { + t.Fatalf("DeleteConversationId: %v", err) + } + + _, err = qr.SelectConversationId(ctx, c.Id) + if err == nil { + t.Fatal("expected error selecting deleted conversation") + } +} + +func testApprovals(t *testing.T, ctx context.Context, db *SQLite) { + var ( + qr = New(db.Reader) + qw = New(db.Writer) + ) + + u, err := qw.InsertUser(ctx, InsertUserParams{ + Name: "approvals", + Password: "test", + Role: UserRoleOwner, + }) + if err != nil { + t.Fatalf("InsertUser: %v", err) + } + + c, err := qw.InsertConversation(ctx, InsertConversationParams{ + UserId: u.Id, + }) + if err != nil { + t.Fatalf("InsertConversation: %v", err) + } + + msg, err := qw.InsertMessage(ctx, InsertMessageParams{ + ConversationId: c.Id, + Sequence: 1, + Role: MessageRoleAssistant, + Content: `{"role":"assistant","content":null,"tool_calls":[{"id":"call_abc","type":"function","function":{"name":"host_run","arguments":"{\"command\":\"rm -rf /\"}"}}]}`, + }) + if err != nil { + t.Fatalf("InsertMessage: %v", err) + } + + // Insert an approval. + approval, err := qw.InsertApproval(ctx, InsertApprovalParams{ + MessageId: msg.Id, + ToolCallId: "call_abc", + }) + if err != nil { + t.Fatalf("InsertApproval: %v", err) + } + if approval.Status != ApprovalStatusPending { + t.Fatalf("expected pending, got %s", approval.Status) + } + if approval.ToolCallId != "call_abc" { + t.Fatalf("expected tool_call_id call_abc, got %s", approval.ToolCallId) + } + + // Select by ID. + res, err := qr.SelectApprovalId(ctx, approval.Id) + if err != nil { + t.Fatalf("SelectApprovalId: %v", err) + } + if res == nil { + t.Fatal("SelectApprovalId returned nil") + } + if res.MessageId != msg.Id { + t.Errorf("expected message_id %s, got %s", msg.Id, res.MessageId) + } + + // Select by message ID. + approvals, err := qr.SelectApprovalsMessageId(ctx, msg.Id) + if err != nil { + t.Fatalf("SelectApprovalsMessageId: %v", err) + } + if len(approvals) != 1 { + t.Fatalf("expected 1 approval, got %d", len(approvals)) + } + + // Select pending. + pending, err := qr.SelectPendingApprovals(ctx) + if err != nil { + t.Fatalf("SelectPendingApprovals: %v", err) + } + if len(pending) != 1 { + t.Fatalf("expected 1 pending approval, got %d", len(pending)) + } + + // Transition to approved. + if err := qw.UpdateApprovalStatus(ctx, UpdateApprovalStatusParams{ + Id: approval.Id, + Status: ApprovalStatusApproved, + }); err != nil { + t.Fatalf("UpdateApprovalStatus: %v", err) + } + + res, err = qr.SelectApprovalId(ctx, approval.Id) + if err != nil { + t.Fatalf("SelectApprovalId after update: %v", err) + } + if res.Status != ApprovalStatusApproved { + t.Fatalf("expected approved, got %s", res.Status) + } + + // Pending should now be empty. + pending, err = qr.SelectPendingApprovals(ctx) + if err != nil { + t.Fatalf("SelectPendingApprovals after update: %v", err) + } + if len(pending) != 0 { + t.Fatalf("expected 0 pending, got %d", len(pending)) + } + + // Insert a second approval and deny it. + approval2, err := qw.InsertApproval(ctx, InsertApprovalParams{ + MessageId: msg.Id, + ToolCallId: "call_def", + }) + if err != nil { + t.Fatalf("InsertApproval second: %v", err) + } + + if err := qw.UpdateApprovalStatus(ctx, UpdateApprovalStatusParams{ + Id: approval2.Id, + Status: ApprovalStatusDenied, + }); err != nil { + t.Fatalf("UpdateApprovalStatus denied: %v", err) + } + + res2, err := qr.SelectApprovalId(ctx, approval2.Id) + if err != nil { + t.Fatalf("SelectApprovalId after deny: %v", err) + } + if res2.Status != ApprovalStatusDenied { + t.Fatalf("expected denied, got %s", res2.Status) + } + + // Both approvals for the message, ordered by created_at. + all, err := qr.SelectApprovalsMessageId(ctx, msg.Id) + if err != nil { + t.Fatalf("SelectApprovalsMessageId final: %v", err) + } + if len(all) != 2 { + t.Fatalf("expected 2 approvals, got %d", len(all)) + } + if all[0].Status != ApprovalStatusApproved { + t.Fatalf("expected first approval approved, got %s", all[0].Status) + } + if all[1].Status != ApprovalStatusDenied { + t.Fatalf("expected second approval denied, got %s", all[1].Status) + } +} diff --git a/internal/db/schema.go b/internal/db/schema.go new file mode 100644 index 0000000..c3dca08 --- /dev/null +++ b/internal/db/schema.go @@ -0,0 +1,171 @@ +package db + +import ( + "context" + "embed" + "fmt" + "log/slog" + "path/filepath" + "sort" + "strconv" + "strings" +) + +//go:embed schema/*.sql +var migrations embed.FS + +// ApplySchema reads the current database version from PRAGMA user_version +// and applies any pending migrations in a single transaction. Each migration +// file runs in order; if any fail the entire transaction rolls back and the +// version is unchanged. +// +// Migration files are named .. The version +// must be a non-negative integer. On success, PRAGMA user_version is set +// to the highest migration version within the same transaction. +// +// If any migration fails the application should refuse to start, since +// the application code expects a schema that matches the latest migration. +func (s *SQLite) ApplySchema(ctx context.Context, log *slog.Logger) error { + if log == nil { + log = slog.Default() + } + + var currentVersion int + if err := s.Writer.QueryRowContext( + ctx, + "PRAGMA user_version", + ).Scan(¤tVersion); err != nil { + return fmt.Errorf("read user_version: %w", err) + } + log.InfoContext( + ctx, + "current schema version", + slog.Int("user_version", currentVersion), + ) + + entries, err := migrations.ReadDir("schema") + if err != nil { + return fmt.Errorf("read schema directory: %w", err) + } + + var pending []struct { + name string + sql []byte + version int + } + for _, e := range entries { + if e.IsDir() { + continue + } + if !strings.HasSuffix(e.Name(), ".sql") { + continue + } + version, err := parseVersion(e.Name()) + if err != nil { + return err + } + if version <= currentVersion { + log.InfoContext( + ctx, + "skipping migration", + slog.String("file", e.Name()), + slog.Int("version", version), + ) + continue + } + sql, err := migrations.ReadFile( + filepath.Join("schema", e.Name()), + ) + if err != nil { + return fmt.Errorf( + "read migration %q: %w", e.Name(), err, + ) + } + pending = append(pending, struct { + name string + sql []byte + version int + }{name: e.Name(), sql: sql, version: version}) + } + sort.Slice(pending, func(i, j int) bool { + return pending[i].version < pending[j].version + }) + if len(pending) == 0 { + log.InfoContext(ctx, "no pending migrations") + return nil + } + log.InfoContext( + ctx, + "pending migrations", + slog.Int("count", len(pending)), + ) + + tx, err := s.Writer.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("begin transaction: %w", err) + } + for _, m := range pending { + if _, err := tx.ExecContext(ctx, string(m.sql)); err != nil { + if err := tx.Rollback(); err != nil { + log.ErrorContext( + ctx, + "rollback failed after apply error", + slog.Any("error", err), + ) + } + return fmt.Errorf( + "apply migration %q (v%d): %w", + m.name, m.version, err, + ) + } + } + + // Set version to the latest migration within the transaction. + latestVersion := pending[len(pending)-1].version + if _, err := tx.ExecContext( + ctx, + fmt.Sprintf("PRAGMA user_version = %d", latestVersion), + ); err != nil { + if err := tx.Rollback(); err != nil { + log.ErrorContext( + ctx, + "rollback failed after version set error", + slog.Any("error", err), + ) + } + return fmt.Errorf( + "set user_version to %d: %w", + latestVersion, err, + ) + } + + if err := tx.Commit(); err != nil { + return fmt.Errorf("commit migrations: %w", err) + } + log.InfoContext( + ctx, + "schema up to date", + slog.Int("user_version", latestVersion), + ) + + return nil +} + +// parseVersion extracts the leading integer from a migration filename +// such as "1.init.sql" -> 1 or "12.add-foo.sql" -> 12. +func parseVersion(name string) (int, error) { + dot := strings.Index(name, ".") + if dot <= 0 { + return 0, fmt.Errorf( + "invalid filename %q: no dot separator", name, + ) + } + v, err := strconv.Atoi(name[:dot]) + if err != nil { + return 0, fmt.Errorf( + "invalid filename %q: version is not an integer: %w", + name, err, + ) + } + return v, nil +} diff --git a/internal/db/schema/1.init.sql b/internal/db/schema/1.init.sql new file mode 100644 index 0000000..3052fab --- /dev/null +++ b/internal/db/schema/1.init.sql @@ -0,0 +1,130 @@ +-- -*- mode: sql; sql-product: sqlite; -*- + +-- This file defines the odidere database schema. +-- +-- Conversations contain messages stored as individual rows. Each +-- message stores its wire-format JSON in the content column, matching +-- the OpenAI Chat Completions API message structure. This avoids +-- per-part decomposition while keeping messages individually addressable +-- for editing, deletion, and branching. +-- +-- Timestamps are stored as TEXT in RFC3339 format. Tables use STRICT typing. + +-- Users are authenticated accounts. The name is the login identifier, +-- unique case-insensitively. The role column is constrained to a fixed set; +-- at this time only the "owner" role is supported. +-- Deleting a user cascades to all their sessions and conversations. +CREATE TABLE IF NOT EXISTS users ( + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), + deactivated_at TEXT, + id TEXT PRIMARY KEY, + name TEXT NOT NULL UNIQUE COLLATE NOCASE, + password TEXT NOT NULL, + role TEXT NOT NULL, + updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), + CHECK (role IN ('owner')) +) STRICT; + +CREATE INDEX IF NOT EXISTS idx_users_deactivated ON users(deactivated_at); + +-- Sessions represent authenticated login sessions. The session id is used +-- as the bearer token in HTTP requests. +CREATE TABLE IF NOT EXISTS sessions ( + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), + deactivated_at TEXT, + id TEXT PRIMARY KEY, + updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), + user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE +) STRICT; + +CREATE INDEX IF NOT EXISTS idx_sessions_active ON sessions(user_id, created_at) WHERE deactivated_at IS NULL; + +-- Conversations are top-level containers for message threads. Each +-- conversation belongs to a single user. Deleting a conversation +-- cascades to all its messages. +CREATE TABLE IF NOT EXISTS conversations ( + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), + id TEXT PRIMARY KEY, + title TEXT, + updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), + user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE +) STRICT; + +CREATE INDEX IF NOT EXISTS idx_conv_user ON conversations(user_id); + +-- Messages are ordered entries within a conversation. The content column +-- stores the wire-format JSON for the message — the exact JSON object +-- that appears in the OpenAI Chat Completions messages array. The role +-- column is extracted for querying without JSON parsing. The sequence +-- column defines ordering within the conversation. Deleting a message +-- cascades to nothing further (all message data is in the row). +CREATE TABLE IF NOT EXISTS messages ( + conversation_id TEXT NOT NULL REFERENCES conversations(id) ON DELETE CASCADE, + content TEXT NOT NULL, + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), + id TEXT PRIMARY KEY, + role TEXT NOT NULL, + sequence INTEGER NOT NULL, + updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), + CHECK (role IN ('system', 'developer', 'user', 'assistant', 'tool')), + UNIQUE(conversation_id, sequence) +) STRICT; + +CREATE INDEX IF NOT EXISTS idx_msg_conv ON messages(conversation_id); +CREATE INDEX IF NOT EXISTS idx_msg_conv_seq ON messages(conversation_id, sequence); + +-- A conversation's timestamp is its synchronization token. +-- Keep it current whenever the conversation message set changes. +CREATE TRIGGER IF NOT EXISTS messages_insert_conversation_updated_at +AFTER INSERT ON messages +BEGIN + UPDATE conversations + SET updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') + WHERE id = NEW.conversation_id; +END; + +CREATE TRIGGER IF NOT EXISTS messages_update_conversation_updated_at +AFTER UPDATE ON messages +BEGIN + UPDATE conversations + SET updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') + WHERE id = NEW.conversation_id; +END; + +CREATE TRIGGER IF NOT EXISTS messages_delete_conversation_updated_at +AFTER DELETE ON messages +BEGIN + UPDATE conversations + SET updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') + WHERE id = OLD.conversation_id; +END; + +-- Approvals track tool calls that require user permission before +-- execution. When the LLM requests a tool call that needs approval, a row +-- is inserted with status "pending" and the agent loop pauses. The UI +-- queries pending approvals to present an inbox to the user. When the +-- user decides, the status transitions to "approved" (proceed) or +-- "denied" (refuse), and the agent loop resumes. +-- +-- Each approval references a specific tool call within an assistant +-- message via message_id and tool_call_id. The tool call's name and +-- arguments are stored in the message's content JSON; they are not +-- duplicated here. The tool's execution result (stdout, stderr, exit +-- code) is stored as a separate tool-role message in the messages +-- table, not in the approval. +CREATE TABLE IF NOT EXISTS approvals ( + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), + id TEXT PRIMARY KEY, + message_id TEXT NOT NULL REFERENCES messages(id) ON DELETE CASCADE, + status TEXT NOT NULL DEFAULT 'pending', + tool_call_id TEXT NOT NULL, + updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), + CHECK (status IN ('pending', 'approved', 'denied')) +) STRICT; + +CREATE INDEX IF NOT EXISTS idx_approvals_message ON approvals(message_id); +CREATE INDEX IF NOT EXISTS idx_approvals_pending ON approvals(status) WHERE status = 'pending'; + +-- Default user. Temporary workaround until authentication is built. +INSERT INTO users (id, name, password, role) +VALUES ('default', 'owner', '', 'owner'); diff --git a/internal/db/schema_test.go b/internal/db/schema_test.go new file mode 100644 index 0000000..6cb99a4 --- /dev/null +++ b/internal/db/schema_test.go @@ -0,0 +1,136 @@ +package db + +import ( + "context" + "testing" +) + +// ExpectedUserVersion is the schema version after all migrations have been applied. +// Increment this when adding a new migration file. +const ExpectedUserVersion = 1 + +func TestSchema(t *testing.T) { + var ( + ctx = t.Context() + db = createDB(t) + ) + defer db.Close() + + t.Run("InitialVersion", func(t *testing.T) { + testInitialVersion(t, ctx, db) + }) + + if err := db.ApplySchema(ctx, nil); err != nil { + t.Fatalf("ApplySchema: %v", err) + } + + t.Run("ApplySchema", func(t *testing.T) { + testApplySchema(t, ctx, db) + }) + t.Run("ApplySchemaIdempotent", func(t *testing.T) { + testSchemaIdempotent(t, ctx, db) + }) + t.Run("ApplySchemaTablesExist", func(t *testing.T) { + testTablesExist(t, ctx, db) + }) + t.Run("ApplySchemaIndexesExist", func(t *testing.T) { + testSchemaIndexes(t, ctx, db) + }) +} + +func testInitialVersion(t *testing.T, ctx context.Context, db *SQLite) { + var version int + err := db.Reader.QueryRowContext( + ctx, + "PRAGMA user_version", + ).Scan(&version) + if err != nil { + t.Fatalf("PRAGMA user_version: %v", err) + } + if version != 0 { + t.Fatalf("expected initial version 0, got %d", version) + } +} + +func testApplySchema(t *testing.T, ctx context.Context, db *SQLite) { + var version int + err := db.Reader.QueryRowContext( + ctx, + "PRAGMA user_version", + ).Scan(&version) + if err != nil { + t.Fatalf("PRAGMA user_version: %v", err) + } + if version != ExpectedUserVersion { + t.Fatalf("expected version %d, got %d", ExpectedUserVersion, version) + } +} + +func testTablesExist(t *testing.T, ctx context.Context, db *SQLite) { + expected := []string{ + "approvals", + "conversations", + "messages", + "sessions", + "users", + } + for _, table := range expected { + var exists bool + err := db.Reader.QueryRowContext( + ctx, + "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ? LIMIT 1)", + table, + ).Scan(&exists) + if err != nil { + t.Fatalf("query sqlite_master for %s: %v", table, err) + } + if !exists { + t.Errorf("table %s does not exist", table) + } + } +} + +func testSchemaIdempotent(t *testing.T, ctx context.Context, db *SQLite) { + // Apply schema again; should succeed without error. + if err := db.ApplySchema(ctx, nil); err != nil { + t.Fatalf("second ApplySchema: %v", err) + } + + var version int + err := db.Reader.QueryRowContext( + ctx, + "PRAGMA user_version", + ).Scan(&version) + if err != nil { + t.Fatalf("PRAGMA user_version: %v", err) + } + if version != ExpectedUserVersion { + t.Fatalf("expected version %d, got %d", ExpectedUserVersion, version) + } +} + +func testSchemaIndexes(t *testing.T, ctx context.Context, db *SQLite) { + expected := []string{ + "idx_approvals_message", + "idx_approvals_pending", + "idx_conv_user", + "idx_msg_conv", + "idx_msg_conv_seq", + "idx_sessions_active", + "idx_users_deactivated", + } + for _, idx := range expected { + var exists bool + err := db.Reader.QueryRowContext( + ctx, + "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'index' AND name = ? LIMIT 1)", + idx, + ).Scan(&exists) + if err != nil { + t.Fatalf("query sqlite_master for index %s: %v", idx, err) + } + if !exists { + t.Errorf("index %s does not exist", idx) + } + } +}