Add database package
This commit is contained in:
@@ -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
|
||||||
|
}
|
||||||
@@ -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")
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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"`
|
||||||
|
}
|
||||||
@@ -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
|
||||||
|
}
|
||||||
@@ -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)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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 <version>.<description>. 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
|
||||||
|
}
|
||||||
@@ -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');
|
||||||
@@ -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)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user