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 updateConversationTitle = ` UPDATE conversations SET title = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') WHERE id = ? RETURNING created_at, id, title, updated_at, user_id ` // UpdateConversationTitleParams holds arguments for // UpdateConversationTitle. type UpdateConversationTitleParams struct { Id string Title string } // UpdateConversationTitle sets the title for a conversation, bumps // updated_at, and returns the updated row. It returns sql.ErrNoRows // if no conversation has that id. func (q *Queries) UpdateConversationTitle( ctx context.Context, arg UpdateConversationTitleParams, ) (*Conversation, error) { row := q.db.QueryRowContext( ctx, updateConversationTitle, arg.Title, arg.Id, ) 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("UpdateConversationTitle: %w", err) } if c.CreatedAt, err = time.Parse(time.RFC3339, createdAt); err != nil { return nil, fmt.Errorf("UpdateConversationTitle: parse created_at: %w", err) } if c.UpdatedAt, err = time.Parse(time.RFC3339, updatedAt); err != nil { return nil, fmt.Errorf("UpdateConversationTitle: parse updated_at: %w", err) } return &c, nil } const updateConversationTitleIfUnset = ` UPDATE conversations SET title = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') WHERE id = ? AND title IS NULL RETURNING created_at, id, title, updated_at, user_id ` // UpdateConversationTitleIfUnsetParams holds arguments for // UpdateConversationTitleIfUnset. type UpdateConversationTitleIfUnsetParams struct { Id string Title string } // UpdateConversationTitleIfUnset sets the first title for a // conversation when title is NULL, bumps updated_at, and returns // the updated row. It prevents a caller from overwriting a title // set by a user or a scheduled job. It returns sql.ErrNoRows if // the conversation does not exist or already has a title; callers // treat both cases as no-op. func (q *Queries) UpdateConversationTitleIfUnset( ctx context.Context, arg UpdateConversationTitleIfUnsetParams, ) (*Conversation, error) { row := q.db.QueryRowContext( ctx, updateConversationTitleIfUnset, arg.Title, arg.Id, ) 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("UpdateConversationTitleIfUnset: %w", err) } if c.CreatedAt, err = time.Parse(time.RFC3339, createdAt); err != nil { return nil, fmt.Errorf("UpdateConversationTitleIfUnset: parse created_at: %w", err) } if c.UpdatedAt, err = time.Parse(time.RFC3339, updatedAt); err != nil { return nil, fmt.Errorf("UpdateConversationTitleIfUnset: parse updated_at: %w", err) } return &c, 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. // Malformed JSON causes insertion to fail. Content jsontext.Value // ConversationId identifies the conversation that owns the new row. ConversationId string // Id is the new row's UUID. Nil generates a version 4 UUID. Id *uuid.UUID // 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 id uuid.UUID if arg.Id == nil { id = uuid.NewV4() } else { id = *arg.Id } var metadataArg any if arg.Metadata != nil { metadataArg = string(arg.Metadata) } row := q.db.QueryRowContext( ctx, insertMessage, id.String(), string(arg.Content), arg.ConversationId, metadataArg, arg.Role, arg.Sequence, ) var ( m Message content []byte metadata []byte createdAt string updatedAt string err error ) if err := row.Scan( &content, &m.ConversationId, &createdAt, &m.Id, &metadata, &m.Role, &m.Sequence, &updatedAt, ); err != nil { return nil, fmt.Errorf("InsertMessage: %w", err) } m.Content = jsontext.Value(content) 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 selectConversationMessageId = ` SELECT content, conversation_id, created_at, id, json(metadata), role, sequence, updated_at FROM messages WHERE conversation_id = ? AND id = ? ` // SelectConversationMessageId returns a conversation's message by ID. func (q *Queries) SelectConversationMessageId( ctx context.Context, conversationId string, id string, ) (*Message, error) { row := q.db.QueryRowContext( ctx, selectConversationMessageId, conversationId, id, ) var ( m Message content []byte metadata []byte createdAt string updatedAt string ) err := row.Scan( &content, &m.ConversationId, &createdAt, &m.Id, &metadata, &m.Role, &m.Sequence, &updatedAt, ) if err != nil { return nil, fmt.Errorf("SelectConversationMessageId: %w", err) } m.Content = jsontext.Value(content) m.Metadata = jsontext.Value(metadata) if m.CreatedAt, err = time.Parse(time.RFC3339, createdAt); err != nil { return nil, fmt.Errorf( "SelectConversationMessageId: parse created_at: %w", err, ) } if m.UpdatedAt, err = time.Parse(time.RFC3339, updatedAt); err != nil { return nil, fmt.Errorf( "SelectConversationMessageId: 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 content []byte metadata []byte createdAt string updatedAt string ) if err := rows.Scan( &content, &m.ConversationId, &createdAt, &m.Id, &metadata, &m.Role, &m.Sequence, &updatedAt, ); err != nil { return nil, fmt.Errorf("SelectMessagesConversationId: %w", err) } m.Content = jsontext.Value(content) 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 = ? RETURNING content, conversation_id, created_at, id, json(metadata), role, sequence, updated_at ` // UpdateMessageContentParams holds arguments for UpdateMessageContent. type UpdateMessageContentParams struct { Content jsontext.Value Id string } // UpdateMessageContent sets the content JSON for a message and returns // the updated row. func (q *Queries) UpdateMessageContent(ctx context.Context, arg UpdateMessageContentParams) (*Message, error) { row := q.db.QueryRowContext( ctx, updateMessageContent, string(arg.Content), arg.Id, ) var ( m Message content []byte metadata []byte createdAt string updatedAt string err error ) if err := row.Scan( &content, &m.ConversationId, &createdAt, &m.Id, &metadata, &m.Role, &m.Sequence, &updatedAt, ); err != nil { return nil, fmt.Errorf("UpdateMessageContent: %w", err) } m.Content = jsontext.Value(content) m.Metadata = jsontext.Value(metadata) if m.CreatedAt, err = time.Parse(time.RFC3339, createdAt); err != nil { return nil, fmt.Errorf("UpdateMessageContent: parse created_at: %w", err) } if m.UpdatedAt, err = time.Parse(time.RFC3339, updatedAt); err != nil { return nil, fmt.Errorf("UpdateMessageContent: parse updated_at: %w", err) } return &m, 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 updateMessagesSequence = ` UPDATE messages SET sequence = sequence - 1, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') WHERE conversation_id = ? AND sequence > ? ` // UpdateMessagesSequence decrements the sequence of every message that // follows the given sequence in the conversation. Every moved row gets a // new updated_at because its position changed. func (q *Queries) UpdateMessagesSequence( ctx context.Context, conversationId string, sequence int, ) error { _, err := q.db.ExecContext( ctx, updateMessagesSequence, conversationId, sequence, ) if err != nil { return fmt.Errorf("UpdateMessagesSequence: %w", err) } return nil } // The shift takes two statements because SQLite checks // UNIQUE(conversation_id, sequence) row by row, not at the end of the // statement. A single "sequence = sequence + 1" fails if it visits row n // while row n+1 still exists. Step one negates the shifted rows to values // no real row uses (sequences start at 1); step two flips them back. const updateMessagesSequenceNegate = ` UPDATE messages SET sequence = -(sequence + 1) WHERE conversation_id = ? AND sequence >= ? ` const updateMessagesSequenceRestore = ` UPDATE messages SET sequence = -sequence, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') WHERE conversation_id = ? AND sequence < 0 ` // UpdateMessagesSequenceIncrement increments the sequence of every message // at or after the given sequence in the conversation, freeing that sequence // for an insert. Every moved row gets a new updated_at because its position // changed. // It runs two statements; call it inside a transaction so a failure // between them cannot leave negative sequences behind. func (q *Queries) UpdateMessagesSequenceIncrement( ctx context.Context, conversationId string, sequence int, ) error { if _, err := q.db.ExecContext( ctx, updateMessagesSequenceNegate, conversationId, sequence, ); err != nil { return fmt.Errorf("UpdateMessagesSequenceIncrement: %w", err) } if _, err := q.db.ExecContext( ctx, updateMessagesSequenceRestore, conversationId, ); err != nil { return fmt.Errorf("UpdateMessagesSequenceIncrement: %w", err) } return nil } // selectMessageSequenceToolResult returns the sequence where a new tool // result belongs: right after the assistant's existing results. The caller // frees that sequence with UpdateMessagesSequenceIncrement, then inserts. // // For example: // // 2 assistant (calls A, B, C) // 3 tool (A) // 4 tool (B) // 5 user // // A result for C belongs at 5; the user message moves to 6. Without the // user message, 5 would be the end of the conversation, and the result // would still go at 5. const selectMessageSequenceToolResult = ` SELECT COALESCE( -- The first message after the assistant that is not a tool result -- (row 5 above) marks the end of the assistant's results. The new -- result takes its place. If there is no such message, MIN returns -- NULL and COALESCE uses the fallback below. MIN(sequence), -- Fallback: the results run to the end of the conversation, so -- append. This is the same calculation as NextSequence. ( SELECT COALESCE(MAX(sequence), 0) + 1 FROM messages WHERE conversation_id = ? ) ) -- MIN always returns one row, even when nothing matches, so the -- caller never gets sql.ErrNoRows. FROM messages WHERE conversation_id = ? AND sequence > ? AND role != 'tool' ` // SelectMessageSequenceToolResult returns the sequence at which to insert // a new tool result for the assistant message at assistantSequence. // See selectMessageSequenceToolResult. func (q *Queries) SelectMessageSequenceToolResult( ctx context.Context, conversationId string, assistantSequence int, ) (int, error) { var sequence int // The subquery's placeholder comes first in the query text. err := q.db.QueryRowContext( ctx, selectMessageSequenceToolResult, conversationId, conversationId, assistantSequence, ).Scan(&sequence) if err != nil { return 0, fmt.Errorf("SelectMessageSequenceToolResult: %w", err) } return sequence, nil } // selectConversationToolResult returns the tool result a call has // among its assistant's results: after the assistant, before the // sequence from selectMessageSequenceToolResult. The search stops there // because tool call IDs are only unique within one assistant message; // another assistant's results may reuse the ID. const selectConversationToolResult = ` SELECT content, conversation_id, created_at, id, json(metadata), role, sequence, updated_at FROM messages WHERE conversation_id = ? AND role = 'tool' AND sequence > ? AND sequence < ? AND json_extract(content, '$.tool_call_id') = ? ` // SelectConversationToolResult returns toolCallId's tool result with a // sequence strictly between after and before. It returns // sql.ErrNoRows if the call has no result in that range. func (q *Queries) SelectConversationToolResult( ctx context.Context, conversationId string, after int, before int, toolCallId string, ) (*Message, error) { row := q.db.QueryRowContext( ctx, selectConversationToolResult, conversationId, after, before, toolCallId, ) var ( m Message content []byte metadata []byte createdAt string updatedAt string ) err := row.Scan( &content, &m.ConversationId, &createdAt, &m.Id, &metadata, &m.Role, &m.Sequence, &updatedAt, ) if err != nil { return nil, fmt.Errorf("SelectConversationToolResult: %w", err) } m.Content = jsontext.Value(content) m.Metadata = jsontext.Value(metadata) if m.CreatedAt, err = time.Parse(time.RFC3339, createdAt); err != nil { return nil, fmt.Errorf( "SelectConversationToolResult: parse created_at: %w", err, ) } if m.UpdatedAt, err = time.Parse(time.RFC3339, updatedAt); err != nil { return nil, fmt.Errorf( "SelectConversationToolResult: parse updated_at: %w", err, ) } return &m, 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 } const deleteExecutionMessageIdToolCallId = ` DELETE FROM executions WHERE message_id = ? AND tool_call_id = ? ` // DeleteExecutionMessageIdToolCallId deletes the execution record for one // tool call within a message. It is a no-op if no record exists. func (q *Queries) DeleteExecutionMessageIdToolCallId( ctx context.Context, messageId string, toolCallId string, ) error { _, err := q.db.ExecContext( ctx, deleteExecutionMessageIdToolCallId, messageId, toolCallId, ) if err != nil { return fmt.Errorf( "DeleteExecutionMessageIdToolCallId: %w", err, ) } return nil } // insertExecutionCompleted marks a call's execution completed, creating // the row if the call never had one. An existing row is completed // regardless of its status, so a pending or denied call is treated as // answered and Inference will not wait on or run it. const insertExecutionCompleted = ` INSERT INTO executions (id, message_id, status, tool_call_id) VALUES (?, ?, 'completed', ?) ON CONFLICT (message_id, tool_call_id) DO UPDATE SET status = 'completed', updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') ` // InsertExecutionCompleted creates or completes the execution for one tool // call within a message. See insertExecutionCompleted. func (q *Queries) InsertExecutionCompleted( ctx context.Context, messageId string, toolCallId string, ) error { _, err := q.db.ExecContext( ctx, insertExecutionCompleted, uuid.New().String(), messageId, toolCallId, ) if err != nil { return fmt.Errorf("InsertExecutionCompleted: %w", err) } return nil }