package db import ( "context" "database/sql" "encoding/json/jsontext" "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 := row.Scan( &createdAt, &u.DeactivatedAt, &u.Id, &u.Name, &u.Password, &u.Role, &updatedAt, ) if 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, metadata, role, sequence) VALUES (?, ?, ?, jsonb(?), ?, ?) RETURNING content, conversation_id, created_at, id, json(metadata), role, sequence, updated_at ` // InsertMessageParams holds arguments for InsertMessage. type InsertMessageParams struct { // Content is the model-visible message JSON saved in the new row. Content string // ConversationId identifies the conversation that owns the new row. ConversationId string // Metadata is inference JSON converted to JSONB for storage. // Nil stores SQL NULL; malformed JSON causes insertion to fail. Metadata jsontext.Value // Role is the message's role, saved separately for row queries. Role MessageRole // Sequence is the new row's ordering position within the conversation. Sequence int } // InsertMessage creates a new message. func (q *Queries) InsertMessage(ctx context.Context, arg InsertMessageParams) (*Message, error) { var metadataArg any if arg.Metadata != nil { metadataArg = string(arg.Metadata) } row := q.db.QueryRowContext( ctx, insertMessage, uuid.New().String(), arg.Content, arg.ConversationId, metadataArg, arg.Role, arg.Sequence, ) var ( m Message metadata []byte createdAt string updatedAt string err error ) if err := row.Scan( &m.Content, &m.ConversationId, &createdAt, &m.Id, &metadata, &m.Role, &m.Sequence, &updatedAt, ); err != nil { return nil, fmt.Errorf("InsertMessage: %w", err) } m.Metadata = jsontext.Value(metadata) 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, json(metadata), 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 metadata []byte createdAt string updatedAt string ) if err := rows.Scan( &m.Content, &m.ConversationId, &createdAt, &m.Id, &metadata, &m.Role, &m.Sequence, &updatedAt, ); err != nil { return nil, fmt.Errorf("SelectMessagesConversationId: %w", err) } m.Metadata = jsontext.Value(metadata) 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 insertExecution = ` INSERT INTO executions (id, message_id, status, tool_call_id) VALUES (?, ?, ?, ?) RETURNING created_at, id, message_id, status, tool_call_id, updated_at ` // InsertExecutionParams holds arguments for InsertExecution. type InsertExecutionParams struct { MessageId string Status ExecutionStatus ToolCallId string } // InsertExecution creates a new execution row. func (q *Queries) InsertExecution(ctx context.Context, arg InsertExecutionParams) (*Execution, error) { row := q.db.QueryRowContext( ctx, insertExecution, uuid.New().String(), arg.MessageId, arg.Status, arg.ToolCallId, ) var ( e Execution createdAt string updatedAt string ) err := row.Scan( &createdAt, &e.Id, &e.MessageId, &e.Status, &e.ToolCallId, &updatedAt, ) if err != nil { return nil, fmt.Errorf( "InsertExecution: %w", err, ) } if e.CreatedAt, err = time.Parse(time.RFC3339, createdAt); err != nil { return nil, fmt.Errorf("InsertExecution: parse created_at: %w", err) } if e.UpdatedAt, err = time.Parse(time.RFC3339, updatedAt); err != nil { return nil, fmt.Errorf("InsertExecution: parse updated_at: %w", err) } return &e, nil } const selectExecutionId = ` SELECT created_at, id, message_id, status, tool_call_id, updated_at FROM executions WHERE id = ? ` // SelectExecutionId returns a single execution row by ID. func (q *Queries) SelectExecutionId(ctx context.Context, id string) (*Execution, error) { row := q.db.QueryRowContext(ctx, selectExecutionId, id) var ( e Execution createdAt string updatedAt string ) err := row.Scan( &createdAt, &e.Id, &e.MessageId, &e.Status, &e.ToolCallId, &updatedAt, ) if err != nil { return nil, fmt.Errorf("SelectExecutionId: %w", err) } if e.CreatedAt, err = time.Parse(time.RFC3339, createdAt); err != nil { return nil, fmt.Errorf("SelectExecutionId: parse created_at: %w", err) } if e.UpdatedAt, err = time.Parse(time.RFC3339, updatedAt); err != nil { return nil, fmt.Errorf("SelectExecutionId: parse updated_at: %w", err) } return &e, nil } const selectExecutionsMessageId = ` SELECT created_at, id, message_id, status, tool_call_id, updated_at FROM executions WHERE message_id = ? ORDER BY created_at ` // SelectExecutionsMessageId returns all execution rows for a message, // ordered by creation time. func (q *Queries) SelectExecutionsMessageId(ctx context.Context, messageId string) ([]Execution, error) { rows, err := q.db.QueryContext(ctx, selectExecutionsMessageId, messageId) if err != nil { return nil, fmt.Errorf("SelectExecutionsMessageId: %w", err) } defer rows.Close() var executions []Execution for rows.Next() { var ( e Execution createdAt string updatedAt string ) if err := rows.Scan( &createdAt, &e.Id, &e.MessageId, &e.Status, &e.ToolCallId, &updatedAt, ); err != nil { return nil, fmt.Errorf("SelectExecutionsMessageId: %w", err) } if e.CreatedAt, err = time.Parse(time.RFC3339, createdAt); err != nil { return nil, fmt.Errorf("SelectExecutionsMessageId: parse created_at: %w", err) } if e.UpdatedAt, err = time.Parse(time.RFC3339, updatedAt); err != nil { return nil, fmt.Errorf("SelectExecutionsMessageId: parse updated_at: %w", err) } executions = append(executions, e) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("SelectExecutionsMessageId: %w", err) } return executions, nil } const selectPendingExecutions = ` SELECT created_at, id, message_id, status, tool_call_id, updated_at FROM executions WHERE status = 'pending' ORDER BY created_at ` // SelectPendingExecutions returns all pending execution rows across all // conversations, ordered by creation time. This serves as the global // inbox for the UI. func (q *Queries) SelectPendingExecutions(ctx context.Context) ([]Execution, error) { rows, err := q.db.QueryContext(ctx, selectPendingExecutions) if err != nil { return nil, fmt.Errorf("SelectPendingExecutions: %w", err) } defer rows.Close() var executions []Execution for rows.Next() { var ( e Execution createdAt string updatedAt string ) if err := rows.Scan( &createdAt, &e.Id, &e.MessageId, &e.Status, &e.ToolCallId, &updatedAt, ); err != nil { return nil, fmt.Errorf("SelectPendingExecutions: %w", err) } if e.CreatedAt, err = time.Parse(time.RFC3339, createdAt); err != nil { return nil, fmt.Errorf("SelectPendingExecutions: parse created_at: %w", err) } if e.UpdatedAt, err = time.Parse(time.RFC3339, updatedAt); err != nil { return nil, fmt.Errorf("SelectPendingExecutions: parse updated_at: %w", err) } executions = append(executions, e) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("SelectPendingExecutions: %w", err) } return executions, nil } const updateExecutionStatus = ` UPDATE executions SET status = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') WHERE id = ? ` // UpdateExecutionStatusParams holds arguments for UpdateExecutionStatus. type UpdateExecutionStatusParams struct { Id string Status ExecutionStatus } // UpdateExecutionStatus transitions an execution row's status and bumps // updated_at. The caller is responsible for valid transitions. func (q *Queries) UpdateExecutionStatus(ctx context.Context, arg UpdateExecutionStatusParams) error { _, err := q.db.ExecContext( ctx, updateExecutionStatus, arg.Status, arg.Id, ) if err != nil { return fmt.Errorf("UpdateExecutionStatus: %w", err) } return nil }