diff --git a/Cargo.lock b/Cargo.lock index abd696bc..bc216374 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6128,6 +6128,7 @@ dependencies = [ name = "nvisy-inference" version = "0.1.0" dependencies = [ + "async-stream", "derive_more", "futures", "rig", diff --git a/crates/nvisy-inference/Cargo.toml b/crates/nvisy-inference/Cargo.toml index a364db2b..7780cabc 100644 --- a/crates/nvisy-inference/Cargo.toml +++ b/crates/nvisy-inference/Cargo.toml @@ -26,6 +26,7 @@ schema = ["dep:schemars"] [dependencies] # Async runtime futures = { workspace = true, features = [] } +async-stream = { workspace = true, features = [] } # LLM framework: provider clients + the classic Agent runtime rig = { workspace = true, features = [] } diff --git a/crates/nvisy-inference/src/client/erased_agent.rs b/crates/nvisy-inference/src/client/erased_agent.rs new file mode 100644 index 00000000..ad3061fd --- /dev/null +++ b/crates/nvisy-inference/src/client/erased_agent.rs @@ -0,0 +1,76 @@ +//! [`ErasedAgent`]: an object-safe view of a rig [`Agent`], erasing the +//! provider's concrete completion-model type so a single handle can hold any +//! backend. + +use futures::future::BoxFuture; +use futures::stream::{BoxStream, StreamExt}; +use rig::agent::{Agent, MultiTurnStreamItem}; +use rig::completion::message::Text; +use rig::completion::{Chat, CompletionModel, GetTokenUsage, Message, Prompt, PromptError}; +use rig::streaming::{StreamedAssistantContent, StreamingChat}; + +use crate::error::Error; + +/// Object-safe view of a rig [`Agent`], erasing the provider's concrete +/// completion-model type so a single handle can hold any backend. +pub(crate) trait ErasedAgent: Send + Sync { + /// Send a single prompt with no prior context. + fn prompt(&self, prompt: String) -> BoxFuture<'_, Result>; + + /// Run one chat turn against `history`, appending the committed messages. + fn chat<'a>( + &'a self, + prompt: String, + history: &'a mut Vec, + ) -> BoxFuture<'a, Result>; + + /// Stream one chat turn against `history` as text deltas. + fn stream_chat<'a>( + &'a self, + prompt: String, + history: Vec, + ) -> BoxFuture<'a, BoxStream<'a, Result>>; +} + +impl ErasedAgent for Agent +where + M: CompletionModel + 'static, + M::StreamingResponse: GetTokenUsage, +{ + fn prompt(&self, prompt: String) -> BoxFuture<'_, Result> { + Box::pin(async move { Prompt::prompt(self, prompt).await }) + } + + fn chat<'a>( + &'a self, + prompt: String, + history: &'a mut Vec, + ) -> BoxFuture<'a, Result> { + Box::pin(async move { Chat::chat(self, prompt, history).await }) + } + + fn stream_chat<'a>( + &'a self, + prompt: String, + history: Vec, + ) -> BoxFuture<'a, BoxStream<'a, Result>> { + Box::pin(async move { + // Map rig's multi-turn stream down to bare text deltas here, inside + // the concrete-`M` impl, so the boxed stream is provider-agnostic and + // the trait stays object-safe. + let stream = StreamingChat::stream_chat(self, prompt, history).await; + let deltas = stream.filter_map(|item| async move { + match item { + Ok(MultiTurnStreamItem::StreamAssistantItem( + StreamedAssistantContent::Text(Text { text, .. }), + )) => Some(Ok(text)), + // Non-text items (tool calls, reasoning, the final response + // marker) carry no user-visible text: drop them. + Ok(_) => None, + Err(err) => Some(Err(Error::Prompt(err.to_string()))), + } + }); + deltas.boxed() + }) + } +} diff --git a/crates/nvisy-inference/src/client/mod.rs b/crates/nvisy-inference/src/client/mod.rs index a6f449ac..89868f62 100644 --- a/crates/nvisy-inference/src/client/mod.rs +++ b/crates/nvisy-inference/src/client/mod.rs @@ -2,55 +2,30 @@ //! //! [`InferenceClient`] is a thin, cloneable wrapper around any provider's rig //! [`Agent`] that provides convenience methods for the most common operations — -//! one-shot [`prompt`](InferenceClient::prompt) and multi-turn -//! [`chat`](InferenceClient::chat). Every public method is instrumented with -//! [`tracing`] for observability. It plays the same role for inference that -//! `ObjectStoreClient` plays for object storage: one runtime handle callers use -//! regardless of which provider backs it. +//! one-shot [`prompt`](InferenceClient::prompt), multi-turn +//! [`chat`](InferenceClient::chat), and streaming +//! [`stream_chat`](InferenceClient::stream_chat). Every public method is +//! instrumented with [`tracing`] for observability. It plays the same role for +//! inference that `ObjectStoreClient` plays for object storage: one runtime +//! handle callers use regardless of which provider backs it. + +mod erased_agent; +mod token_stream; +mod turn; use std::sync::Arc; -use futures::future::BoxFuture; +use async_stream::stream; +use futures::StreamExt; use rig::agent::Agent; use rig::client::verify::{VerifyClient, VerifyError}; -/// A single conversation message, re-exported so callers can build the -/// [`chat`](InferenceClient::chat) history without depending on `rig` directly. -pub use rig::completion::Message; -use rig::completion::{Chat, CompletionModel, Prompt, PromptError}; +use rig::completion::{CompletionModel, GetTokenUsage, Message}; +use self::erased_agent::ErasedAgent; +pub use self::token_stream::TokenStream; +pub use self::turn::{ChatTurn, Role}; use crate::error::Error; -/// Object-safe view of a rig [`Agent`], erasing the provider's concrete -/// completion-model type so a single handle can hold any backend. -trait ErasedAgent: Send + Sync { - /// Send a single prompt with no prior context. - fn prompt(&self, prompt: String) -> BoxFuture<'_, Result>; - - /// Run one chat turn against `history`, appending the committed messages. - fn chat<'a>( - &'a self, - prompt: String, - history: &'a mut Vec, - ) -> BoxFuture<'a, Result>; -} - -impl ErasedAgent for Agent -where - M: CompletionModel + 'static, -{ - fn prompt(&self, prompt: String) -> BoxFuture<'_, Result> { - Box::pin(async move { Prompt::prompt(self, prompt).await }) - } - - fn chat<'a>( - &'a self, - prompt: String, - history: &'a mut Vec, - ) -> BoxFuture<'a, Result> { - Box::pin(async move { Chat::chat(self, prompt, history).await }) - } -} - /// Cloneable handle to any inference backend (OpenAI, Anthropic, Ollama, ...). /// /// Wraps a provider's rig agent behind a provider-agnostic interface, so callers @@ -63,6 +38,7 @@ impl InferenceClient { pub(crate) fn new(agent: Agent) -> Self where M: CompletionModel + 'static, + M::StreamingResponse: GetTokenUsage, { Self(Arc::new(agent)) } @@ -79,16 +55,46 @@ impl InferenceClient { /// Run one chat turn against `history`, returning the model's text response. /// - /// `history` is caller-owned and updated in place: the prompt and the - /// messages the model commits this turn are appended to it, so passing the - /// same `Vec` across calls continues the conversation. + /// `history` is the prior conversation as [`ChatTurn`]s. #[tracing::instrument(name = "inference.chat", skip_all, fields(history_len = history.len()))] - pub async fn chat(&self, prompt: &str, history: &mut Vec) -> Result { + pub async fn chat(&self, prompt: &str, history: Vec) -> Result { + let mut history = to_messages(history); self.0 - .chat(prompt.to_owned(), history) + .chat(prompt.to_owned(), &mut history) .await .map_err(|err| Error::Prompt(err.to_string())) } + + /// Stream one chat turn against `history`, yielding the model's response as + /// text deltas. + /// + /// Returns immediately with a [`TokenStream`]; the request opens lazily when + /// the stream is first polled. Each item is a token chunk as it arrives, and + /// a failure mid-generation ends the stream with an `Err`. `history` is the + /// prior conversation as [`ChatTurn`]s; persist the user prompt and the + /// assembled reply on the caller side. + #[tracing::instrument(name = "inference.stream_chat", skip_all, fields(history_len = history.len()))] + pub fn stream_chat(&self, prompt: &str, history: Vec) -> TokenStream { + // Own the agent handle in the generator so the result is `'static` and + // can outlive this `InferenceClient` (moved into a response body). The + // provider stream borrows the owned `Arc`, which the coroutine keeps + // alive for the whole stream. + let agent = Arc::clone(&self.0); + let prompt = prompt.to_owned(); + let history = to_messages(history); + let inner = stream! { + let mut deltas = agent.stream_chat(prompt, history).await; + while let Some(delta) = deltas.next().await { + yield delta; + } + }; + TokenStream::new(inner.boxed()) + } +} + +/// Converts a provider-agnostic history into rig messages. +fn to_messages(history: Vec) -> Vec { + history.into_iter().map(Message::from).collect() } /// Verifies a built provider client's credentials against the provider. diff --git a/crates/nvisy-inference/src/client/token_stream.rs b/crates/nvisy-inference/src/client/token_stream.rs new file mode 100644 index 00000000..2b4c962d --- /dev/null +++ b/crates/nvisy-inference/src/client/token_stream.rs @@ -0,0 +1,37 @@ +//! The [`TokenStream`] type: an assistant response streamed as text deltas. + +use std::pin::Pin; +use std::task::{Context, Poll}; + +use futures::Stream; +use futures::stream::{BoxStream, StreamExt}; + +use crate::error::Error; + +/// A stream of the assistant's response as text deltas. +/// +/// Each item is a token chunk (`Ok`); a failure mid-generation is the final item +/// (`Err`) and ends the stream. Owned and `'static`, so it can be moved into a +/// response body outliving the client that produced it. +/// +/// Yielded by [`InferenceClient::stream_chat`](crate::InferenceClient::stream_chat). +/// Poll it with the [`Stream`] API ([`futures::StreamExt`]). +#[must_use = "a token stream does nothing unless polled"] +pub struct TokenStream { + inner: BoxStream<'static, Result>, +} + +impl TokenStream { + /// Wraps an owned delta stream. + pub(crate) fn new(inner: BoxStream<'static, Result>) -> Self { + Self { inner } + } +} + +impl Stream for TokenStream { + type Item = Result; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + self.inner.poll_next_unpin(cx) + } +} diff --git a/crates/nvisy-inference/src/client/turn.rs b/crates/nvisy-inference/src/client/turn.rs new file mode 100644 index 00000000..49d42827 --- /dev/null +++ b/crates/nvisy-inference/src/client/turn.rs @@ -0,0 +1,60 @@ +//! The provider-agnostic conversation types callers build history from. +//! +//! These wrap the underlying rig message model so consumers depend on this +//! crate's own types rather than `rig`'s. + +use rig::completion::Message; + +/// Who authored a conversation turn. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Role { + /// A system instruction. + System, + /// A message from the user. + User, + /// A reply from the assistant. + Assistant, +} + +/// One turn of a conversation: a role and its text. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ChatTurn { + /// Who authored the turn. + pub role: Role, + /// The turn's text. + pub content: String, +} + +impl ChatTurn { + /// A system turn. + pub fn system(content: impl Into) -> Self { + Self::of(Role::System, content) + } + + /// A user turn. + pub fn user(content: impl Into) -> Self { + Self::of(Role::User, content) + } + + /// An assistant turn. + pub fn assistant(content: impl Into) -> Self { + Self::of(Role::Assistant, content) + } + + fn of(role: Role, content: impl Into) -> Self { + Self { + role, + content: content.into(), + } + } +} + +impl From for Message { + fn from(turn: ChatTurn) -> Self { + match turn.role { + Role::System => Message::system(turn.content), + Role::User => Message::user(turn.content), + Role::Assistant => Message::assistant(turn.content), + } + } +} diff --git a/crates/nvisy-inference/src/lib.rs b/crates/nvisy-inference/src/lib.rs index 0c407441..070a9724 100644 --- a/crates/nvisy-inference/src/lib.rs +++ b/crates/nvisy-inference/src/lib.rs @@ -2,8 +2,9 @@ #![cfg_attr(docsrs, feature(doc_cfg))] #![doc = include_str!("../README.md")] -pub mod client; +mod client; mod error; pub mod providers; +pub use client::{ChatTurn, InferenceClient, Role, TokenStream, verify}; pub use error::Error; diff --git a/crates/nvisy-postgres/src/model/chat_message.rs b/crates/nvisy-postgres/src/model/chat_message.rs new file mode 100644 index 00000000..bd1825c2 --- /dev/null +++ b/crates/nvisy-postgres/src/model/chat_message.rs @@ -0,0 +1,42 @@ +//! Chat message model for PostgreSQL database operations. + +use diesel::prelude::*; +use jiff_diesel::Timestamp; +use uuid::Uuid; + +use crate::schema::chat_messages; +use crate::types::ChatRole; + +/// One message in a chat session. +#[derive(Debug, Clone, PartialEq, Queryable, Selectable)] +#[diesel(table_name = chat_messages)] +#[diesel(check_for_backend(diesel::pg::Pg))] +pub struct ChatMessage { + /// Unique message identifier. + pub id: Uuid, + /// Session this message belongs to. + pub session_id: Uuid, + /// Parent in the conversation tree; `None` is a root. + pub parent_id: Option, + /// Author of the message. + pub role: ChatRole, + /// Message text, XChaCha20-Poly1305 encrypted with the workspace key. + pub content: Vec, + /// Message creation timestamp. + pub created_at: Timestamp, +} + +/// Data for appending a new chat message. +#[derive(Debug, Clone, Insertable)] +#[diesel(table_name = chat_messages)] +#[diesel(check_for_backend(diesel::pg::Pg))] +pub struct NewChatMessage { + /// Session this message belongs to. + pub session_id: Uuid, + /// Parent in the conversation tree; `None` is a root. + pub parent_id: Option, + /// Author of the message. + pub role: ChatRole, + /// Message text, XChaCha20-Poly1305 encrypted with the workspace key. + pub content: Vec, +} diff --git a/crates/nvisy-postgres/src/model/chat_session.rs b/crates/nvisy-postgres/src/model/chat_session.rs new file mode 100644 index 00000000..fe847797 --- /dev/null +++ b/crates/nvisy-postgres/src/model/chat_session.rs @@ -0,0 +1,56 @@ +//! Chat session model for PostgreSQL database operations. + +use diesel::prelude::*; +use jiff_diesel::Timestamp; +use uuid::Uuid; + +use crate::schema::chat_sessions; + +/// A workspace-scoped assistant conversation thread. +#[derive(Debug, Clone, PartialEq, Queryable, Selectable)] +#[diesel(table_name = chat_sessions)] +#[diesel(check_for_backend(diesel::pg::Pg))] +pub struct ChatSession { + /// Unique session identifier. + pub id: Uuid, + /// Workspace this session belongs to. + pub workspace_id: Uuid, + /// Account that opened the session. + pub account_id: Uuid, + /// Human-readable title, seeded from the first message. + pub title: String, + /// Active leaf of the message tree (the conversation's resume point). + pub current_message_id: Option, + /// Session creation timestamp. + pub created_at: Timestamp, + /// Timestamp of the most recent message. + pub updated_at: Timestamp, + /// Soft-deletion timestamp; `None` means live. + pub deleted_at: Option, +} + +/// Data for creating a new chat session. +#[derive(Debug, Clone, Insertable)] +#[diesel(table_name = chat_sessions)] +#[diesel(check_for_backend(diesel::pg::Pg))] +pub struct NewChatSession { + /// Workspace this session belongs to. + pub workspace_id: Uuid, + /// Account that opened the session. + pub account_id: Uuid, + /// Human-readable title. + pub title: String, +} + +/// Data for updating a chat session. +#[derive(Debug, Default, Clone, AsChangeset)] +#[diesel(table_name = chat_sessions)] +#[diesel(check_for_backend(diesel::pg::Pg))] +pub struct UpdateChatSession { + /// New title. + pub title: Option, + /// New active leaf of the message tree. + pub current_message_id: Option>, + /// New most-recent-activity timestamp. + pub updated_at: Option, +} diff --git a/crates/nvisy-postgres/src/model/mod.rs b/crates/nvisy-postgres/src/model/mod.rs index 8f31f7e2..da9f821e 100644 --- a/crates/nvisy-postgres/src/model/mod.rs +++ b/crates/nvisy-postgres/src/model/mod.rs @@ -6,6 +6,8 @@ mod account; mod account_api_token; mod account_notification; +mod chat_message; +mod chat_session; mod pipeline_reference; mod workspace; mod workspace_activity; @@ -27,6 +29,9 @@ pub use account_api_token::{AccountApiToken, NewAccountApiToken, UpdateAccountAp pub use account_notification::{ AccountNotification, NewAccountNotification, UpdateAccountNotification, }; +// Chat models +pub use chat_message::{ChatMessage, NewChatMessage}; +pub use chat_session::{ChatSession, NewChatSession, UpdateChatSession}; pub use pipeline_reference::PipelinePolicy; // Workspace models pub use workspace::{NewWorkspace, UpdateWorkspace, Workspace}; diff --git a/crates/nvisy-postgres/src/model/workspace_connection.rs b/crates/nvisy-postgres/src/model/workspace_connection.rs index 20fd092c..1410052c 100644 --- a/crates/nvisy-postgres/src/model/workspace_connection.rs +++ b/crates/nvisy-postgres/src/model/workspace_connection.rs @@ -6,7 +6,7 @@ use serde_json::Value as JsonValue; use uuid::Uuid; use crate::schema::workspace_connections; -use crate::types::{HasCreatedAt, HasDeletedAt, HasUpdatedAt}; +use crate::types::{HasCreatedAt, HasDeletedAt, HasUpdatedAt, ProviderType}; /// Workspace connection model: a generic encrypted provider connection. /// @@ -28,6 +28,8 @@ pub struct WorkspaceConnection { pub display_name: String, /// Provider identifier (`s3`, `azure`, `gcs`, `openai`, `ollama`, ...). pub provider: String, + /// Capability category of the provider (object store, language model, ...). + pub provider_type: ProviderType, /// Encrypted connection config (XChaCha20-Poly1305 encrypted JSON): /// provider tag, credentials, and any provider-specific settings. pub encrypted_data: Vec, @@ -56,6 +58,8 @@ pub struct NewWorkspaceConnection { pub display_name: String, /// Provider identifier, for indexing and filtering. pub provider: String, + /// Capability category of the provider. + pub provider_type: ProviderType, /// Encrypted connection config. pub encrypted_data: Vec, /// Whether the connection is enabled. diff --git a/crates/nvisy-postgres/src/query/chat_message.rs b/crates/nvisy-postgres/src/query/chat_message.rs new file mode 100644 index 00000000..1e7e62b3 --- /dev/null +++ b/crates/nvisy-postgres/src/query/chat_message.rs @@ -0,0 +1,143 @@ +//! Chat messages repository. + +use std::future::Future; + +use diesel::prelude::*; +use diesel_async::RunQueryDsl; +use uuid::Uuid; + +use crate::model::{ChatMessage, NewChatMessage, UpdateChatSession}; +use crate::{PgConnection, PgError, PgResult, schema}; + +/// What to update on a message's session when appending it, applied in the same +/// transaction as the insert so the message and its session state never diverge. +#[derive(Debug, Clone, Default)] +pub struct AppendSessionUpdate { + /// Point the session's active leaf at the newly appended message. + pub advance_leaf: bool, + /// Set the session's title (e.g. seeded from the first message). + pub title: Option, +} + +/// Repository for chat message database operations. +pub trait ChatMessageRepository { + /// Appends a message and updates its session per `session_update` in one + /// transaction, so the session's active leaf and title never diverge from + /// its messages. The session's `updated_at` is always bumped. Returns the + /// stored message. + fn append_chat_message( + &mut self, + new_message: NewChatMessage, + session_update: AppendSessionUpdate, + ) -> impl Future> + Send; + + /// Loads all of a session's messages (the whole tree), oldest first. + fn list_chat_messages( + &mut self, + session_id: Uuid, + ) -> impl Future>> + Send; + + /// Finds a message by id within a session, scoping a client-supplied parent + /// to the session it belongs to. + fn find_chat_message_in_session( + &mut self, + session_id: Uuid, + message_id: Uuid, + ) -> impl Future>> + Send; +} + +impl ChatMessageRepository for PgConnection { + async fn append_chat_message( + &mut self, + new_message: NewChatMessage, + session_update: AppendSessionUpdate, + ) -> PgResult { + use diesel::dsl::now; + use diesel_async::AsyncConnection; + use schema::{chat_messages, chat_sessions}; + + let session_id = new_message.session_id; + + // Insert the message and update its session atomically, so the active + // leaf and title never diverge from the messages. `updated_at` is always + // bumped so the session-list ordering reflects the latest message. The + // active leaf is set to the row just inserted (its id is known only here). + self.transaction(async |conn| { + let message = diesel::insert_into(chat_messages::table) + .values(&new_message) + .returning(ChatMessage::as_returning()) + .get_result(conn) + .await + .map_err(PgError::from)?; + + let update = UpdateChatSession { + title: session_update.title, + current_message_id: session_update.advance_leaf.then_some(Some(message.id)), + updated_at: None, + }; + diesel::update(chat_sessions::table.filter(chat_sessions::id.eq(session_id))) + .set((update, chat_sessions::updated_at.eq(now))) + .execute(conn) + .await + .map_err(PgError::from)?; + + Ok::<_, PgError>(message) + }) + .await + } + + async fn list_chat_messages(&mut self, session_id: Uuid) -> PgResult> { + use schema::chat_messages::{self, dsl}; + + chat_messages::table + .filter(dsl::session_id.eq(session_id)) + .order(dsl::created_at.asc()) + .select(ChatMessage::as_select()) + .load(self) + .await + .map_err(PgError::from) + } + + async fn find_chat_message_in_session( + &mut self, + session_id: Uuid, + message_id: Uuid, + ) -> PgResult> { + use schema::chat_messages::{self, dsl}; + + chat_messages::table + .filter(dsl::id.eq(message_id)) + .filter(dsl::session_id.eq(session_id)) + .select(ChatMessage::as_select()) + .first(self) + .await + .optional() + .map_err(PgError::from) + } +} + +impl ChatMessage { + /// The active conversation path ending at `leaf_id`: the chain of messages + /// from the root down to that leaf, in chronological order. + /// + /// Follows `parent_id` links up from the leaf through `messages` (the + /// session's full message set), then reverses. A `None` leaf, or a leaf not + /// present, yields an empty path. Sessions are small, so walking the loaded + /// set in memory is cheaper and simpler than a recursive query. + #[must_use] + pub fn path_to(messages: &[ChatMessage], leaf_id: Option) -> Vec<&ChatMessage> { + use std::collections::HashMap; + + let by_id: HashMap = messages.iter().map(|m| (m.id, m)).collect(); + + let mut path = Vec::new(); + let mut cursor = leaf_id; + while let Some(id) = cursor { + let Some(message) = by_id.get(&id) else { break }; + path.push(*message); + cursor = message.parent_id; + } + path.reverse(); + path + } +} diff --git a/crates/nvisy-postgres/src/query/chat_session.rs b/crates/nvisy-postgres/src/query/chat_session.rs new file mode 100644 index 00000000..442e6fcb --- /dev/null +++ b/crates/nvisy-postgres/src/query/chat_session.rs @@ -0,0 +1,136 @@ +//! Chat sessions repository. + +use std::future::Future; + +use diesel::prelude::*; +use diesel_async::RunQueryDsl; +use uuid::Uuid; + +use crate::model::{ChatSession, NewChatSession, UpdateChatSession}; +use crate::types::OffsetPagination; +use crate::{PgConnection, PgError, PgResult, schema}; + +/// Repository for chat session database operations. +pub trait ChatSessionRepository { + /// Creates a new chat session. + fn create_chat_session( + &mut self, + new_session: NewChatSession, + ) -> impl Future> + Send; + + /// Finds a live session by id within a workspace. + fn find_chat_session_in_workspace( + &mut self, + workspace_id: Uuid, + session_id: Uuid, + ) -> impl Future>> + Send; + + /// Lists a workspace's live sessions, most recently active first. + fn list_chat_sessions( + &mut self, + workspace_id: Uuid, + pagination: OffsetPagination, + ) -> impl Future>> + Send; + + /// Updates a session (title and/or activity timestamp). + fn update_chat_session( + &mut self, + session_id: Uuid, + updates: UpdateChatSession, + ) -> impl Future> + Send; + + /// Soft-deletes a live session within a workspace, returning whether a live + /// session was deleted. + fn delete_chat_session( + &mut self, + workspace_id: Uuid, + session_id: Uuid, + ) -> impl Future> + Send; +} + +impl ChatSessionRepository for PgConnection { + async fn create_chat_session(&mut self, new_session: NewChatSession) -> PgResult { + use schema::chat_sessions; + + diesel::insert_into(chat_sessions::table) + .values(&new_session) + .returning(ChatSession::as_returning()) + .get_result(self) + .await + .map_err(PgError::from) + } + + async fn find_chat_session_in_workspace( + &mut self, + workspace_id: Uuid, + session_id: Uuid, + ) -> PgResult> { + use schema::chat_sessions::{self, dsl}; + + chat_sessions::table + .filter(dsl::id.eq(session_id)) + .filter(dsl::workspace_id.eq(workspace_id)) + .filter(dsl::deleted_at.is_null()) + .select(ChatSession::as_select()) + .first(self) + .await + .optional() + .map_err(PgError::from) + } + + async fn list_chat_sessions( + &mut self, + workspace_id: Uuid, + pagination: OffsetPagination, + ) -> PgResult> { + use schema::chat_sessions::{self, dsl}; + + chat_sessions::table + .filter(dsl::workspace_id.eq(workspace_id)) + .filter(dsl::deleted_at.is_null()) + .order(dsl::updated_at.desc()) + .limit(pagination.limit) + .offset(pagination.offset) + .select(ChatSession::as_select()) + .load(self) + .await + .map_err(PgError::from) + } + + async fn update_chat_session( + &mut self, + session_id: Uuid, + updates: UpdateChatSession, + ) -> PgResult { + use schema::chat_sessions::{self, dsl}; + + diesel::update(chat_sessions::table.filter(dsl::id.eq(session_id))) + .set(updates) + .returning(ChatSession::as_returning()) + .get_result(self) + .await + .map_err(PgError::from) + } + + async fn delete_chat_session( + &mut self, + workspace_id: Uuid, + session_id: Uuid, + ) -> PgResult { + use diesel::dsl::now; + use schema::chat_sessions::{self, dsl}; + + let affected = diesel::update( + chat_sessions::table + .filter(dsl::id.eq(session_id)) + .filter(dsl::workspace_id.eq(workspace_id)) + .filter(dsl::deleted_at.is_null()), + ) + .set(dsl::deleted_at.eq(now)) + .execute(self) + .await + .map_err(PgError::from)?; + + Ok(affected > 0) + } +} diff --git a/crates/nvisy-postgres/src/query/mod.rs b/crates/nvisy-postgres/src/query/mod.rs index 80e3cb28..76ae9831 100644 --- a/crates/nvisy-postgres/src/query/mod.rs +++ b/crates/nvisy-postgres/src/query/mod.rs @@ -16,6 +16,8 @@ mod account; mod account_api_token; mod account_notification; +mod chat_message; +mod chat_session; mod pipeline_reference; mod search; mod workspace; @@ -34,6 +36,8 @@ mod workspace_webhook; pub use account::AccountRepository; pub use account_api_token::AccountApiTokenRepository; pub use account_notification::AccountNotificationRepository; +pub use chat_message::{AppendSessionUpdate, ChatMessageRepository}; +pub use chat_session::ChatSessionRepository; pub use pipeline_reference::PipelineReferenceRepository; pub use workspace::WorkspaceRepository; pub use workspace_activity::WorkspaceActivityRepository; diff --git a/crates/nvisy-postgres/src/query/workspace_connection.rs b/crates/nvisy-postgres/src/query/workspace_connection.rs index 7e0df16d..d9d973e3 100644 --- a/crates/nvisy-postgres/src/query/workspace_connection.rs +++ b/crates/nvisy-postgres/src/query/workspace_connection.rs @@ -8,7 +8,8 @@ use uuid::Uuid; use crate::model::{NewWorkspaceConnection, UpdateWorkspaceConnection, WorkspaceConnection}; use crate::types::{ - AccountRefRow, CursorPage, CursorPagination, OffsetPagination, SyncMode, WithAccountRef, + AccountRefRow, CursorPage, CursorPagination, OffsetPagination, ProviderType, SyncMode, + WithAccountRef, }; use crate::{PgConnection, PgError, PgResult, schema}; @@ -55,6 +56,19 @@ pub trait WorkspaceConnectionRepository { provider: &str, ) -> impl Future>> + Send; + /// Finds the workspace's most recently updated live, enabled connection of a + /// given capability (e.g. its language model), if any. Resolves a capability + /// connection without decrypting every connection's config. + /// + /// Disabled (`is_active = false`) connections are excluded: a disabled + /// connection is not usable, and a newer disabled one must not shadow an + /// active one. + fn find_connection_by_type( + &mut self, + workspace_id: Uuid, + provider_type: ProviderType, + ) -> impl Future>> + Send; + /// Lists all active, import-mode connections that have a sync schedule, /// across every workspace. Used by the scheduled-sync worker. fn list_scheduled_connections( @@ -211,6 +225,26 @@ impl WorkspaceConnectionRepository for PgConnection { Ok(connections) } + async fn find_connection_by_type( + &mut self, + workspace_id: Uuid, + provider_type: ProviderType, + ) -> PgResult> { + use schema::workspace_connections::{self, dsl}; + + workspace_connections::table + .filter(dsl::workspace_id.eq(workspace_id)) + .filter(dsl::provider_type.eq(provider_type)) + .filter(dsl::deleted_at.is_null()) + .filter(dsl::is_active.eq(true)) + .order(dsl::updated_at.desc()) + .select(WorkspaceConnection::as_select()) + .first(self) + .await + .optional() + .map_err(PgError::from) + } + async fn list_scheduled_connections(&mut self) -> PgResult> { use schema::workspace_connection_schedule as sched; use schema::workspace_connections::{self, dsl}; diff --git a/crates/nvisy-postgres/src/schema.rs b/crates/nvisy-postgres/src/schema.rs index 0c6bc1d2..c878140c 100644 --- a/crates/nvisy-postgres/src/schema.rs +++ b/crates/nvisy-postgres/src/schema.rs @@ -9,6 +9,10 @@ pub mod sql_types { #[diesel(postgres_type(name = "api_token_type"))] pub struct ApiTokenType; + #[derive(diesel::query_builder::QueryId, diesel::sql_types::SqlType)] + #[diesel(postgres_type(name = "chat_role"))] + pub struct ChatRole; + #[derive(diesel::query_builder::QueryId, diesel::sql_types::SqlType)] #[diesel(postgres_type(name = "file_kind"))] pub struct FileKind; @@ -33,6 +37,10 @@ pub mod sql_types { #[diesel(postgres_type(name = "pipeline_trigger_type"))] pub struct PipelineTriggerType; + #[derive(diesel::query_builder::QueryId, diesel::sql_types::SqlType)] + #[diesel(postgres_type(name = "provider_type"))] + pub struct ProviderType; + #[derive(diesel::query_builder::QueryId, diesel::sql_types::SqlType)] #[diesel(postgres_type(name = "sync_deletion_policy"))] pub struct SyncDeletionPolicy; @@ -118,6 +126,35 @@ diesel::table! { } } +diesel::table! { + use diesel::sql_types::*; + use super::sql_types::ChatRole; + + chat_messages (id) { + id -> Uuid, + session_id -> Uuid, + parent_id -> Nullable, + role -> ChatRole, + content -> Bytea, + created_at -> Timestamptz, + } +} + +diesel::table! { + use diesel::sql_types::*; + + chat_sessions (id) { + id -> Uuid, + workspace_id -> Uuid, + account_id -> Uuid, + title -> Text, + current_message_id -> Nullable, + created_at -> Timestamptz, + updated_at -> Timestamptz, + deleted_at -> Nullable, + } +} + diesel::table! { use diesel::sql_types::*; use super::sql_types::ActivityType; @@ -169,6 +206,7 @@ diesel::table! { diesel::table! { use diesel::sql_types::*; + use super::sql_types::ProviderType; workspace_connections (id) { id -> Uuid, @@ -176,6 +214,7 @@ diesel::table! { account_id -> Uuid, display_name -> Text, provider -> Text, + provider_type -> ProviderType, encrypted_data -> Bytea, is_active -> Bool, metadata -> Jsonb, @@ -379,6 +418,8 @@ diesel::table! { diesel::joinable!(account_api_tokens -> accounts (account_id)); diesel::joinable!(account_notifications -> accounts (account_id)); +diesel::joinable!(chat_sessions -> accounts (account_id)); +diesel::joinable!(chat_sessions -> workspaces (workspace_id)); diesel::joinable!(workspace_activities -> accounts (account_id)); diesel::joinable!(workspace_activities -> workspaces (workspace_id)); diesel::joinable!(workspace_connection_schedule -> workspace_connections (connection_id)); @@ -407,6 +448,8 @@ diesel::allow_tables_to_appear_in_same_query!( account_api_tokens, account_notifications, accounts, + chat_messages, + chat_sessions, workspace_activities, workspace_connection_schedule, workspace_connection_syncs, diff --git a/crates/nvisy-postgres/src/types/constraint/account_api_tokens.rs b/crates/nvisy-postgres/src/types/constraint/account_api_tokens.rs index 9a3cfae9..3b8ba3bf 100644 --- a/crates/nvisy-postgres/src/types/constraint/account_api_tokens.rs +++ b/crates/nvisy-postgres/src/types/constraint/account_api_tokens.rs @@ -1,62 +1,12 @@ //! Account API tokens table constraint violations. -use serde::{Deserialize, Serialize}; -use strum::{Display, EnumIter, EnumString}; - -use super::ConstraintCategory; +use strum::EnumString; /// Account API tokens table constraint violations. -#[derive(Debug, Clone, Copy, Eq, PartialEq)] -#[derive(Serialize, Deserialize, Display, EnumIter, EnumString)] -#[serde(into = "String", try_from = "String")] +#[derive(Debug, Clone, Copy, Eq, PartialEq, EnumString)] pub enum AccountApiTokenConstraints { - // Token validation constraints #[strum(serialize = "account_api_tokens_display_name_not_empty")] NameNotEmpty, #[strum(serialize = "account_api_tokens_display_name_length")] NameLength, - - // Token chronological constraints - #[strum(serialize = "account_api_tokens_expired_after_issued")] - ExpiredAfterIssued, - #[strum(serialize = "account_api_tokens_deleted_after_issued")] - DeletedAfterIssued, - #[strum(serialize = "account_api_tokens_last_used_after_issued")] - LastUsedAfterIssued, -} - -impl AccountApiTokenConstraints { - /// Creates a new [`AccountApiTokenConstraints`] from the constraint name. - pub fn new(constraint: &str) -> Option { - constraint.parse().ok() - } - - /// Returns the category of this constraint violation. - pub fn categorize(&self) -> ConstraintCategory { - match self { - AccountApiTokenConstraints::NameNotEmpty | AccountApiTokenConstraints::NameLength => { - ConstraintCategory::Validation - } - - AccountApiTokenConstraints::ExpiredAfterIssued - | AccountApiTokenConstraints::DeletedAfterIssued - | AccountApiTokenConstraints::LastUsedAfterIssued => ConstraintCategory::Chronological, - } - } -} - -impl From for String { - #[inline] - fn from(val: AccountApiTokenConstraints) -> Self { - val.to_string() - } -} - -impl TryFrom for AccountApiTokenConstraints { - type Error = strum::ParseError; - - #[inline] - fn try_from(value: String) -> Result { - value.parse() - } } diff --git a/crates/nvisy-postgres/src/types/constraint/account_notifications.rs b/crates/nvisy-postgres/src/types/constraint/account_notifications.rs index 60cda10e..f7fd5098 100644 --- a/crates/nvisy-postgres/src/types/constraint/account_notifications.rs +++ b/crates/nvisy-postgres/src/types/constraint/account_notifications.rs @@ -1,55 +1,10 @@ //! Account notifications table constraint violations. -use serde::{Deserialize, Serialize}; -use strum::{Display, EnumIter, EnumString}; - -use super::ConstraintCategory; +use strum::EnumString; /// Account notifications table constraint violations. -#[derive(Debug, Clone, Copy, Eq, PartialEq)] -#[derive(Serialize, Deserialize, Display, EnumIter, EnumString)] -#[serde(into = "String", try_from = "String")] +#[derive(Debug, Clone, Copy, Eq, PartialEq, EnumString)] pub enum AccountNotificationConstraints { - // Notification validation constraints #[strum(serialize = "account_notifications_params_size")] ParamsSize, - - // Notification chronological constraints - #[strum(serialize = "account_notifications_expires_after_created")] - ExpiresAfterCreated, - #[strum(serialize = "account_notifications_read_after_created")] - ReadAfterCreated, -} - -impl AccountNotificationConstraints { - /// Creates a new [`AccountNotificationConstraints`] from the constraint name. - pub fn new(constraint: &str) -> Option { - constraint.parse().ok() - } - - /// Returns the category of this constraint violation. - pub fn categorize(&self) -> ConstraintCategory { - match self { - AccountNotificationConstraints::ParamsSize => ConstraintCategory::Validation, - - AccountNotificationConstraints::ExpiresAfterCreated - | AccountNotificationConstraints::ReadAfterCreated => ConstraintCategory::Chronological, - } - } -} - -impl From for String { - #[inline] - fn from(val: AccountNotificationConstraints) -> Self { - val.to_string() - } -} - -impl TryFrom for AccountNotificationConstraints { - type Error = strum::ParseError; - - #[inline] - fn try_from(value: String) -> Result { - value.parse() - } } diff --git a/crates/nvisy-postgres/src/types/constraint/accounts.rs b/crates/nvisy-postgres/src/types/constraint/accounts.rs index d57e9078..ec4b24ef 100644 --- a/crates/nvisy-postgres/src/types/constraint/accounts.rs +++ b/crates/nvisy-postgres/src/types/constraint/accounts.rs @@ -1,16 +1,10 @@ //! Accounts table constraint violations. -use serde::{Deserialize, Serialize}; -use strum::{Display, EnumIter, EnumString}; - -use super::ConstraintCategory; +use strum::EnumString; /// Account table constraint violations. -#[derive(Debug, Clone, Copy, Eq, PartialEq)] -#[derive(Serialize, Deserialize, Display, EnumIter, EnumString)] -#[serde(into = "String", try_from = "String")] +#[derive(Debug, Clone, Copy, Eq, PartialEq, EnumString)] pub enum AccountConstraints { - // Account validation constraints #[strum(serialize = "accounts_username_length")] UsernameLength, #[strum(serialize = "accounts_username_format")] @@ -33,69 +27,8 @@ pub enum AccountConstraints { LocaleFormat, #[strum(serialize = "accounts_suspended_not_admin")] SuspendedNotAdmin, - - // Account uniqueness constraints #[strum(serialize = "accounts_username_unique_idx")] UsernameUnique, #[strum(serialize = "accounts_email_address_unique_idx")] EmailUnique, - - // Account chronological constraints - #[strum(serialize = "accounts_updated_after_created")] - UpdatedAfterCreated, - #[strum(serialize = "accounts_deleted_after_created")] - DeletedAfterCreated, - #[strum(serialize = "accounts_deleted_after_updated")] - DeletedAfterUpdated, - #[strum(serialize = "accounts_password_changed_after_created")] - PasswordChangedAfterCreated, -} - -impl AccountConstraints { - /// Creates a new [`AccountConstraints`] from the constraint name. - pub fn new(constraint: &str) -> Option { - constraint.parse().ok() - } - - /// Returns the category of this constraint violation. - pub fn categorize(&self) -> ConstraintCategory { - match self { - AccountConstraints::UsernameLength - | AccountConstraints::UsernameFormat - | AccountConstraints::DisplayNameLength - | AccountConstraints::DisplayNameNotEmpty - | AccountConstraints::EmailFormat - | AccountConstraints::EmailLengthMax - | AccountConstraints::PasswordHashNotEmpty - | AccountConstraints::PasswordHashLengthMin - | AccountConstraints::TimezoneFormat - | AccountConstraints::LocaleFormat - | AccountConstraints::SuspendedNotAdmin => ConstraintCategory::Validation, - - AccountConstraints::UsernameUnique | AccountConstraints::EmailUnique => { - ConstraintCategory::Uniqueness - } - - AccountConstraints::UpdatedAfterCreated - | AccountConstraints::DeletedAfterCreated - | AccountConstraints::DeletedAfterUpdated - | AccountConstraints::PasswordChangedAfterCreated => ConstraintCategory::Chronological, - } - } -} - -impl From for String { - #[inline] - fn from(val: AccountConstraints) -> Self { - val.to_string() - } -} - -impl TryFrom for AccountConstraints { - type Error = strum::ParseError; - - #[inline] - fn try_from(value: String) -> Result { - value.parse() - } } diff --git a/crates/nvisy-postgres/src/types/constraint/chat_messages.rs b/crates/nvisy-postgres/src/types/constraint/chat_messages.rs new file mode 100644 index 00000000..16aa8b35 --- /dev/null +++ b/crates/nvisy-postgres/src/types/constraint/chat_messages.rs @@ -0,0 +1,16 @@ +//! Chat messages table constraint violations. + +use strum::EnumString; + +/// Chat messages table constraint violations. +#[derive(Debug, Clone, Copy, Eq, PartialEq, EnumString)] +pub enum ChatMessageConstraints { + #[strum(serialize = "chat_messages_content_size")] + ContentSize, + + // Tree integrity: a parent must be in the same session. + #[strum(serialize = "chat_messages_id_session_key")] + IdSession, + #[strum(serialize = "chat_messages_parent_fkey")] + Parent, +} diff --git a/crates/nvisy-postgres/src/types/constraint/chat_sessions.rs b/crates/nvisy-postgres/src/types/constraint/chat_sessions.rs new file mode 100644 index 00000000..b2eabab1 --- /dev/null +++ b/crates/nvisy-postgres/src/types/constraint/chat_sessions.rs @@ -0,0 +1,14 @@ +//! Chat sessions table constraint violations. + +use strum::EnumString; + +/// Chat sessions table constraint violations. +/// +/// Enumerates the constraints a client request can trip that map to a specific +/// non-500 response. Server-controlled invariants (ownership and active-leaf +/// foreign keys, timestamp ordering) fall through to the generic handler. +#[derive(Debug, Clone, Copy, Eq, PartialEq, EnumString)] +pub enum ChatSessionConstraints { + #[strum(serialize = "chat_sessions_title_length")] + TitleLength, +} diff --git a/crates/nvisy-postgres/src/types/constraint/files.rs b/crates/nvisy-postgres/src/types/constraint/files.rs index 165dce85..213e2026 100644 --- a/crates/nvisy-postgres/src/types/constraint/files.rs +++ b/crates/nvisy-postgres/src/types/constraint/files.rs @@ -1,100 +1,24 @@ //! Files table constraint violations. -use serde::{Deserialize, Serialize}; -use strum::{Display, EnumIter, EnumString}; - -use super::ConstraintCategory; +use strum::EnumString; /// Files table constraint violations. -#[derive(Debug, Clone, Copy, Eq, PartialEq)] -#[derive(Serialize, Deserialize, Display, EnumIter, EnumString)] -#[serde(into = "String", try_from = "String")] +#[derive(Debug, Clone, Copy, Eq, PartialEq, EnumString)] pub enum WorkspaceFileConstraints { - // File identity validation constraints #[strum(serialize = "workspace_files_display_name_length")] DisplayNameLength, #[strum(serialize = "workspace_files_original_filename_length")] OriginalFilenameLength, #[strum(serialize = "workspace_files_file_extension_format")] FileExtensionFormat, - - // File storage constraints #[strum(serialize = "workspace_files_file_size_min")] FileSizeMin, - #[strum(serialize = "workspace_files_storage_path_not_empty")] - StoragePathNotEmpty, - #[strum(serialize = "workspace_files_storage_bucket_not_empty")] - StorageBucketNotEmpty, - #[strum(serialize = "workspace_files_file_hash_sha256_length")] - FileHashSha256Length, - - // File metadata constraints #[strum(serialize = "workspace_files_metadata_size")] MetadataSize, - - // File version constraints #[strum(serialize = "workspace_files_version_number_min")] VersionNumberMin, - - // Uniqueness constraints #[strum(serialize = "workspace_files_workspace_id_id_key")] WorkspaceIdIdUnique, #[strum(serialize = "workspace_files_source_object_unique_idx")] SourceObjectUnique, - - // File chronological constraints - #[strum(serialize = "workspace_files_updated_after_created")] - UpdatedAfterCreated, - #[strum(serialize = "workspace_files_deleted_after_created")] - DeletedAfterCreated, - #[strum(serialize = "workspace_files_deleted_after_updated")] - DeletedAfterUpdated, - #[strum(serialize = "workspace_files_expires_after_created")] - ExpiresAfterCreated, -} - -impl WorkspaceFileConstraints { - /// Creates a new [`WorkspaceFileConstraints`] from the constraint name. - pub fn new(constraint: &str) -> Option { - constraint.parse().ok() - } - - /// Returns the category of this constraint violation. - pub fn categorize(&self) -> ConstraintCategory { - match self { - WorkspaceFileConstraints::DisplayNameLength - | WorkspaceFileConstraints::OriginalFilenameLength - | WorkspaceFileConstraints::FileExtensionFormat - | WorkspaceFileConstraints::FileSizeMin - | WorkspaceFileConstraints::StoragePathNotEmpty - | WorkspaceFileConstraints::StorageBucketNotEmpty - | WorkspaceFileConstraints::FileHashSha256Length - | WorkspaceFileConstraints::MetadataSize - | WorkspaceFileConstraints::VersionNumberMin => ConstraintCategory::Validation, - - WorkspaceFileConstraints::WorkspaceIdIdUnique - | WorkspaceFileConstraints::SourceObjectUnique => ConstraintCategory::Uniqueness, - - WorkspaceFileConstraints::UpdatedAfterCreated - | WorkspaceFileConstraints::DeletedAfterCreated - | WorkspaceFileConstraints::DeletedAfterUpdated - | WorkspaceFileConstraints::ExpiresAfterCreated => ConstraintCategory::Chronological, - } - } -} - -impl From for String { - #[inline] - fn from(val: WorkspaceFileConstraints) -> Self { - val.to_string() - } -} - -impl TryFrom for WorkspaceFileConstraints { - type Error = strum::ParseError; - - #[inline] - fn try_from(value: String) -> Result { - value.parse() - } } diff --git a/crates/nvisy-postgres/src/types/constraint/mod.rs b/crates/nvisy-postgres/src/types/constraint/mod.rs index d65b2233..dbfbed56 100644 --- a/crates/nvisy-postgres/src/types/constraint/mod.rs +++ b/crates/nvisy-postgres/src/types/constraint/mod.rs @@ -8,6 +8,10 @@ mod account_api_tokens; mod account_notifications; mod accounts; +// Chat constraint modules +mod chat_messages; +mod chat_sessions; + // Workspace-related constraint modules mod workspace_activities; mod workspace_invites; @@ -27,13 +31,11 @@ mod workspace_connection_syncs; mod workspace_connections; mod workspace_policies; -use std::fmt; - -use serde::{Deserialize, Serialize}; - pub use self::account_api_tokens::AccountApiTokenConstraints; pub use self::account_notifications::AccountNotificationConstraints; pub use self::accounts::AccountConstraints; +pub use self::chat_messages::ChatMessageConstraints; +pub use self::chat_sessions::ChatSessionConstraints; pub use self::files::WorkspaceFileConstraints; pub use self::pipeline_references::WorkspacePipelineReferenceConstraints; pub use self::pipeline_runs::WorkspacePipelineRunConstraints; @@ -52,14 +54,17 @@ pub use self::workspaces::WorkspaceConstraints; /// This enum wraps all specific constraint types, providing a single interface /// for handling any constraint violation while maintaining type safety and /// organizational benefits of the separate modules. -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -#[serde(into = "String", try_from = "String")] +#[derive(Debug, Clone, PartialEq, Eq)] pub enum ConstraintViolation { // Account-related constraints Account(AccountConstraints), AccountNotification(AccountNotificationConstraints), AccountApiToken(AccountApiTokenConstraints), + // Chat-related constraints + ChatSession(ChatSessionConstraints), + ChatMessage(ChatMessageConstraints), + // Workspace-related constraints Workspace(WorkspaceConstraints), WorkspaceMember(WorkspaceMemberConstraints), @@ -79,22 +84,6 @@ pub enum ConstraintViolation { WorkspacePolicy(WorkspacePolicyConstraints), } -/// Categories of database constraint violations. -/// -/// This enum helps classify constraint violations by their purpose and type, -/// making it easier to handle different categories of errors appropriately. -#[derive(Debug, Copy, Clone, PartialEq, Eq, Serialize, Deserialize)] -pub enum ConstraintCategory { - /// Data validation constraints (format, length, range checks). - Validation, - /// Chronological integrity constraints (timestamp relationships). - Chronological, - /// Business logic constraints (domain-specific rules). - BusinessLogic, - /// Uniqueness constraints (primary keys, unique indexes). - Uniqueness, -} - impl ConstraintViolation { /// Creates a new [`ConstraintViolation`] from the constraint name. /// @@ -122,170 +111,36 @@ impl ConstraintViolation { /// assert!(unknown.is_none()); /// ``` pub fn new(constraint: &str) -> Option { - let prefix = constraint.split('_').next()?; + // Every per-table enum matches the full constraint name via strum, so + // parsing is tried against each in turn until one succeeds. macro_rules! try_parse { - ($($parser:expr => $variant:ident),+ $(,)?) => { - None$(.or_else(|| $parser(constraint).map(Self::$variant)))+ + ($($variant:ident),+ $(,)?) => { + None$(.or_else(|| constraint.parse().ok().map(Self::$variant)))+ }; } - match prefix { - "accounts" => try_parse!(AccountConstraints::new => Account), - "account" => try_parse! { - AccountNotificationConstraints::new => AccountNotification, - AccountApiTokenConstraints::new => AccountApiToken, - }, - "workspaces" => try_parse!(WorkspaceConstraints::new => Workspace), - // Every workspace-owned table is prefixed `workspace_*`, so all of - // their constraints dispatch here. strum matches the full name, so - // the order of these parsers does not matter. - "workspace" => try_parse! { - WorkspaceMemberConstraints::new => WorkspaceMember, - WorkspaceInviteConstraints::new => WorkspaceInvite, - WorkspaceActivitiesConstraints::new => WorkspaceActivityLog, - WorkspaceWebhookConstraints::new => WorkspaceWebhook, - WorkspaceConnectionSyncConstraints::new => WorkspaceConnectionSync, - WorkspaceConnectionConstraints::new => WorkspaceConnection, - WorkspacePolicyConstraints::new => WorkspacePolicy, - WorkspaceFileConstraints::new => WorkspaceFile, - WorkspacePipelineRunConstraints::new => WorkspacePipelineRun, - WorkspacePipelineConstraints::new => WorkspacePipeline, - WorkspacePipelineReferenceConstraints::new => WorkspacePipelineReference, - }, - _ => None, - } - } - - /// Returns the table name associated with this constraint. - /// - /// This is useful for categorizing errors by the table they affect. - pub fn table_name(&self) -> &'static str { - match self { - // Account-related tables - ConstraintViolation::Account(_) => "accounts", - ConstraintViolation::AccountNotification(_) => "account_notifications", - ConstraintViolation::AccountApiToken(_) => "account_api_tokens", - - // Workspace-related tables - ConstraintViolation::Workspace(_) => "workspaces", - ConstraintViolation::WorkspaceMember(_) => "workspace_members", - ConstraintViolation::WorkspaceInvite(_) => "workspace_invites", - ConstraintViolation::WorkspaceActivityLog(_) => "workspace_activities", - ConstraintViolation::WorkspaceWebhook(_) => "workspace_webhooks", - - // File-related tables - ConstraintViolation::WorkspaceFile(_) => "workspace_files", - - // Pipeline-related tables - ConstraintViolation::WorkspacePipeline(_) => "workspace_pipelines", - ConstraintViolation::WorkspacePipelineRun(_) => "workspace_pipeline_runs", - ConstraintViolation::WorkspacePipelineReference(_) => "pipeline_references", - ConstraintViolation::WorkspaceConnection(_) => "workspace_connections", - ConstraintViolation::WorkspaceConnectionSync(_) => "workspace_connection_syncs", - ConstraintViolation::WorkspacePolicy(_) => "workspace_policies", - } - } - - /// Returns the functional area this constraint belongs to. - /// - /// This groups constraints by their business domain for higher-level categorization. - pub fn functional_area(&self) -> &'static str { - match self { - ConstraintViolation::Account(_) - | ConstraintViolation::AccountNotification(_) - | ConstraintViolation::AccountApiToken(_) => "accounts", - - ConstraintViolation::Workspace(_) - | ConstraintViolation::WorkspaceMember(_) - | ConstraintViolation::WorkspaceInvite(_) - | ConstraintViolation::WorkspaceActivityLog(_) - | ConstraintViolation::WorkspaceWebhook(_) => "workspaces", - - ConstraintViolation::WorkspaceFile(_) => "files", - - ConstraintViolation::WorkspacePipeline(_) - | ConstraintViolation::WorkspacePipelineRun(_) - | ConstraintViolation::WorkspacePipelineReference(_) => "pipelines", - - ConstraintViolation::WorkspaceConnection(_) - | ConstraintViolation::WorkspaceConnectionSync(_) => "connections", - ConstraintViolation::WorkspacePolicy(_) => "policies", - } - } - - /// Returns the category of this constraint violation. - /// - /// This helps categorize errors by their type for better error handling and reporting. - pub fn constraint_category(&self) -> ConstraintCategory { - match self { - ConstraintViolation::Account(c) => c.categorize(), - ConstraintViolation::AccountNotification(c) => c.categorize(), - ConstraintViolation::AccountApiToken(c) => c.categorize(), - - ConstraintViolation::Workspace(c) => c.categorize(), - ConstraintViolation::WorkspaceMember(c) => c.categorize(), - ConstraintViolation::WorkspaceInvite(c) => c.categorize(), - ConstraintViolation::WorkspaceActivityLog(c) => c.categorize(), - ConstraintViolation::WorkspaceWebhook(c) => c.categorize(), - - ConstraintViolation::WorkspaceFile(c) => c.categorize(), - - ConstraintViolation::WorkspacePipeline(c) => c.categorize(), - ConstraintViolation::WorkspacePipelineRun(c) => c.categorize(), - ConstraintViolation::WorkspacePipelineReference(c) => c.categorize(), - ConstraintViolation::WorkspaceConnection(c) => c.categorize(), - ConstraintViolation::WorkspaceConnectionSync(c) => c.categorize(), - ConstraintViolation::WorkspacePolicy(c) => c.categorize(), - } - } - - /// Returns the underlying constraint name as used in the database. - #[inline] - pub fn constraint_name(&self) -> String { - self.to_string() - } -} - -impl fmt::Display for ConstraintViolation { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - match self { - ConstraintViolation::Account(c) => write!(f, "{}", c), - ConstraintViolation::AccountNotification(c) => write!(f, "{}", c), - ConstraintViolation::AccountApiToken(c) => write!(f, "{}", c), - - ConstraintViolation::Workspace(c) => write!(f, "{}", c), - ConstraintViolation::WorkspaceMember(c) => write!(f, "{}", c), - ConstraintViolation::WorkspaceInvite(c) => write!(f, "{}", c), - ConstraintViolation::WorkspaceActivityLog(c) => write!(f, "{}", c), - ConstraintViolation::WorkspaceWebhook(c) => write!(f, "{}", c), - - ConstraintViolation::WorkspaceFile(c) => write!(f, "{}", c), - - ConstraintViolation::WorkspacePipeline(c) => write!(f, "{}", c), - ConstraintViolation::WorkspacePipelineRun(c) => write!(f, "{}", c), - ConstraintViolation::WorkspacePipelineReference(c) => write!(f, "{}", c), - ConstraintViolation::WorkspaceConnection(c) => write!(f, "{}", c), - ConstraintViolation::WorkspaceConnectionSync(c) => write!(f, "{}", c), - ConstraintViolation::WorkspacePolicy(c) => write!(f, "{}", c), + try_parse! { + Account, + AccountNotification, + AccountApiToken, + ChatSession, + ChatMessage, + Workspace, + WorkspaceMember, + WorkspaceInvite, + WorkspaceActivityLog, + WorkspaceWebhook, + WorkspaceFile, + WorkspacePipeline, + WorkspacePipelineRun, + WorkspacePipelineReference, + WorkspaceConnection, + WorkspaceConnectionSync, + WorkspacePolicy, } } } -impl From for String { - #[inline] - fn from(val: ConstraintViolation) -> Self { - val.to_string() - } -} - -impl TryFrom for ConstraintViolation { - type Error = String; - - fn try_from(value: String) -> Result { - Self::new(&value).ok_or_else(|| format!("Unknown constraint: {}", value)) - } -} - #[cfg(test)] mod tests { use super::*; @@ -299,90 +154,13 @@ mod tests { )) ); - // Workspace-owned tables all share the `workspace_*` prefix and dispatch - // through the same arm; the file/pipeline enums live there too. assert_eq!( ConstraintViolation::new("workspace_files_version_number_min"), Some(ConstraintViolation::WorkspaceFile( WorkspaceFileConstraints::VersionNumberMin )) ); - assert_eq!( - ConstraintViolation::new("workspace_pipelines_display_name_length"), - Some(ConstraintViolation::WorkspacePipeline( - WorkspacePipelineConstraints::NameLength - )) - ); - assert_eq!( - ConstraintViolation::new("workspace_policies_workspace_id_id_key"), - Some(ConstraintViolation::WorkspacePolicy( - WorkspacePolicyConstraints::WorkspaceIdIdUnique - )) - ); - assert_eq!(ConstraintViolation::new("unknown_constraint"), None); - } - - #[test] - fn test_table_name_extraction() { - let violation = ConstraintViolation::Account(AccountConstraints::EmailFormat); - assert_eq!(violation.table_name(), "accounts"); - - let violation = ConstraintViolation::Workspace(WorkspaceConstraints::DisplayNameLength); - assert_eq!(violation.table_name(), "workspaces"); - - let violation = - ConstraintViolation::WorkspaceFile(WorkspaceFileConstraints::StoragePathNotEmpty); - assert_eq!(violation.table_name(), "workspace_files"); - - let violation = - ConstraintViolation::WorkspacePolicy(WorkspacePolicyConstraints::NameLength); - assert_eq!(violation.table_name(), "workspace_policies"); - } - - #[test] - fn test_functional_area_extraction() { - let violation = ConstraintViolation::Account(AccountConstraints::EmailFormat); - assert_eq!(violation.functional_area(), "accounts"); - - let violation = - ConstraintViolation::WorkspaceFile(WorkspaceFileConstraints::VersionNumberMin); - assert_eq!(violation.functional_area(), "files"); - - let violation = - ConstraintViolation::WorkspacePolicy(WorkspacePolicyConstraints::NameLength); - assert_eq!(violation.functional_area(), "policies"); - } - - #[test] - fn test_constraint_categorization() { - let violation = ConstraintViolation::Account(AccountConstraints::DisplayNameLength); - assert_eq!( - violation.constraint_category(), - ConstraintCategory::Validation - ); - - let violation = ConstraintViolation::Account(AccountConstraints::UpdatedAfterCreated); - assert_eq!( - violation.constraint_category(), - ConstraintCategory::Chronological - ); - - let violation = ConstraintViolation::WorkspaceConnection( - WorkspaceConnectionConstraints::WorkspaceIdIdUnique, - ); - assert_eq!( - violation.constraint_category(), - ConstraintCategory::Uniqueness - ); - } - #[test] - fn test_constraint_name_method() { - let violation = - ConstraintViolation::WorkspaceFile(WorkspaceFileConstraints::WorkspaceIdIdUnique); - assert_eq!( - violation.constraint_name(), - "workspace_files_workspace_id_id_key" - ); + assert_eq!(ConstraintViolation::new("unknown_constraint"), None); } } diff --git a/crates/nvisy-postgres/src/types/constraint/pipeline_references.rs b/crates/nvisy-postgres/src/types/constraint/pipeline_references.rs index 973e3d5d..4e503c9a 100644 --- a/crates/nvisy-postgres/src/types/constraint/pipeline_references.rs +++ b/crates/nvisy-postgres/src/types/constraint/pipeline_references.rs @@ -1,49 +1,13 @@ //! Pipeline reference join-table constraint violations. -use serde::{Deserialize, Serialize}; -use strum::{Display, EnumIter, EnumString}; - -use super::ConstraintCategory; +use strum::EnumString; /// Foreign-key violations on the pipeline → policy join table. /// /// These fire when a pipeline references a policy id that does not exist in its /// workspace, so they map to a client error rather than a 500. -#[derive(Debug, Clone, Copy, Eq, PartialEq)] -#[derive(Serialize, Deserialize, Display, EnumIter, EnumString)] -#[serde(into = "String", try_from = "String")] +#[derive(Debug, Clone, Copy, Eq, PartialEq, EnumString)] pub enum WorkspacePipelineReferenceConstraints { - // Foreign-key constraints (referenced row must exist in the workspace) #[strum(serialize = "workspace_pipeline_policies_policy_fkey")] PolicyReference, - #[strum(serialize = "workspace_pipeline_policies_pipeline_fkey")] - PolicyPipelineReference, -} - -impl WorkspacePipelineReferenceConstraints { - /// Creates a new [`WorkspacePipelineReferenceConstraints`] from the constraint name. - pub fn new(constraint: &str) -> Option { - constraint.parse().ok() - } - - /// Returns the category of this constraint violation. - pub fn categorize(&self) -> ConstraintCategory { - ConstraintCategory::BusinessLogic - } -} - -impl From for String { - #[inline] - fn from(val: WorkspacePipelineReferenceConstraints) -> Self { - val.to_string() - } -} - -impl TryFrom for WorkspacePipelineReferenceConstraints { - type Error = strum::ParseError; - - #[inline] - fn try_from(value: String) -> Result { - value.parse() - } } diff --git a/crates/nvisy-postgres/src/types/constraint/pipeline_runs.rs b/crates/nvisy-postgres/src/types/constraint/pipeline_runs.rs index 10f6011c..6775b56e 100644 --- a/crates/nvisy-postgres/src/types/constraint/pipeline_runs.rs +++ b/crates/nvisy-postgres/src/types/constraint/pipeline_runs.rs @@ -1,65 +1,14 @@ //! Pipeline runs table constraint violations. -use serde::{Deserialize, Serialize}; -use strum::{Display, EnumIter, EnumString}; - -use super::ConstraintCategory; +use strum::EnumString; /// Pipeline runs table constraint violations. -#[derive(Debug, Clone, Copy, Eq, PartialEq)] -#[derive(Serialize, Deserialize, Display, EnumIter, EnumString)] -#[serde(into = "String", try_from = "String")] +#[derive(Debug, Clone, Copy, Eq, PartialEq, EnumString)] pub enum WorkspacePipelineRunConstraints { - // Size / validation constraints #[strum(serialize = "workspace_pipeline_runs_metadata_size")] MetadataSize, #[strum(serialize = "workspace_pipeline_runs_idempotency_key_length")] IdempotencyKeyLength, - - // Uniqueness constraints #[strum(serialize = "workspace_pipeline_runs_idempotency_idx")] IdempotencyUnique, - - // Chronological constraints - #[strum(serialize = "workspace_pipeline_runs_completed_after_started")] - CompletedAfterStarted, -} - -impl WorkspacePipelineRunConstraints { - /// Creates a new [`WorkspacePipelineRunConstraints`] from the constraint name. - pub fn new(constraint: &str) -> Option { - constraint.parse().ok() - } - - /// Returns the category of this constraint violation. - pub fn categorize(&self) -> ConstraintCategory { - match self { - WorkspacePipelineRunConstraints::MetadataSize - | WorkspacePipelineRunConstraints::IdempotencyKeyLength => { - ConstraintCategory::Validation - } - - WorkspacePipelineRunConstraints::IdempotencyUnique => ConstraintCategory::Uniqueness, - - WorkspacePipelineRunConstraints::CompletedAfterStarted => { - ConstraintCategory::Chronological - } - } - } -} - -impl From for String { - #[inline] - fn from(val: WorkspacePipelineRunConstraints) -> Self { - val.to_string() - } -} - -impl TryFrom for WorkspacePipelineRunConstraints { - type Error = strum::ParseError; - - #[inline] - fn try_from(value: String) -> Result { - value.parse() - } } diff --git a/crates/nvisy-postgres/src/types/constraint/pipelines.rs b/crates/nvisy-postgres/src/types/constraint/pipelines.rs index 8ae62ad7..79aca47b 100644 --- a/crates/nvisy-postgres/src/types/constraint/pipelines.rs +++ b/crates/nvisy-postgres/src/types/constraint/pipelines.rs @@ -1,100 +1,30 @@ //! Pipelines table constraint violations. -use serde::{Deserialize, Serialize}; -use strum::{Display, EnumIter, EnumString}; - -use super::ConstraintCategory; +use strum::EnumString; /// Pipelines table constraint violations. -#[derive(Debug, Clone, Copy, Eq, PartialEq)] -#[derive(Serialize, Deserialize, Display, EnumIter, EnumString)] -#[serde(into = "String", try_from = "String")] +#[derive(Debug, Clone, Copy, Eq, PartialEq, EnumString)] pub enum WorkspacePipelineConstraints { - // Pipeline slug validation constraints #[strum(serialize = "workspace_pipelines_slug_length")] SlugLength, #[strum(serialize = "workspace_pipelines_slug_format")] SlugFormat, - - // Pipeline name validation constraints #[strum(serialize = "workspace_pipelines_display_name_length")] NameLength, - - // Pipeline description validation constraints #[strum(serialize = "workspace_pipelines_description_length")] DescriptionLength, - - // Pipeline definition constraints #[strum(serialize = "workspace_pipelines_definition_size")] DefinitionSize, - - // Pipeline metadata constraints #[strum(serialize = "workspace_pipelines_metadata_size")] MetadataSize, - - // Pipeline schedule validation constraints #[strum(serialize = "workspace_pipelines_schedule_cron_length")] ScheduleCronLength, #[strum(serialize = "workspace_pipelines_schedule_requires_cron")] ScheduleRequiresCron, #[strum(serialize = "workspace_pipelines_schedule_tz_length")] ScheduleTzLength, - - // Uniqueness constraints #[strum(serialize = "workspace_pipelines_workspace_id_id_key")] WorkspaceIdIdUnique, #[strum(serialize = "workspace_pipelines_slug_unique_idx")] SlugUnique, - - // Pipeline chronological constraints - #[strum(serialize = "workspace_pipelines_updated_after_created")] - UpdatedAfterCreated, - #[strum(serialize = "workspace_pipelines_deleted_after_created")] - DeletedAfterCreated, -} - -impl WorkspacePipelineConstraints { - /// Creates a new [`WorkspacePipelineConstraints`] from the constraint name. - pub fn new(constraint: &str) -> Option { - constraint.parse().ok() - } - - /// Returns the category of this constraint violation. - pub fn categorize(&self) -> ConstraintCategory { - match self { - WorkspacePipelineConstraints::SlugLength - | WorkspacePipelineConstraints::SlugFormat - | WorkspacePipelineConstraints::NameLength - | WorkspacePipelineConstraints::DescriptionLength - | WorkspacePipelineConstraints::DefinitionSize - | WorkspacePipelineConstraints::MetadataSize - | WorkspacePipelineConstraints::ScheduleCronLength - | WorkspacePipelineConstraints::ScheduleRequiresCron - | WorkspacePipelineConstraints::ScheduleTzLength => ConstraintCategory::Validation, - - WorkspacePipelineConstraints::WorkspaceIdIdUnique - | WorkspacePipelineConstraints::SlugUnique => ConstraintCategory::Uniqueness, - - WorkspacePipelineConstraints::UpdatedAfterCreated - | WorkspacePipelineConstraints::DeletedAfterCreated => { - ConstraintCategory::Chronological - } - } - } -} - -impl From for String { - #[inline] - fn from(val: WorkspacePipelineConstraints) -> Self { - val.to_string() - } -} - -impl TryFrom for WorkspacePipelineConstraints { - type Error = strum::ParseError; - - #[inline] - fn try_from(value: String) -> Result { - value.parse() - } } diff --git a/crates/nvisy-postgres/src/types/constraint/workspace_activities.rs b/crates/nvisy-postgres/src/types/constraint/workspace_activities.rs index 085931d4..be92dac8 100644 --- a/crates/nvisy-postgres/src/types/constraint/workspace_activities.rs +++ b/crates/nvisy-postgres/src/types/constraint/workspace_activities.rs @@ -1,46 +1,10 @@ //! Workspace activities table constraint violations. -use serde::{Deserialize, Serialize}; -use strum::{Display, EnumIter, EnumString}; - -use super::ConstraintCategory; +use strum::EnumString; /// Workspace activities table constraint violations. -#[derive(Debug, Clone, Copy, Eq, PartialEq)] -#[derive(Serialize, Deserialize, Display, EnumIter, EnumString)] -#[serde(into = "String", try_from = "String")] +#[derive(Debug, Clone, Copy, Eq, PartialEq, EnumString)] pub enum WorkspaceActivitiesConstraints { - // Activity validation constraints #[strum(serialize = "workspace_activities_params_size")] ParamsSize, } - -impl WorkspaceActivitiesConstraints { - /// Creates a new [`WorkspaceActivitiesConstraints`] from the constraint name. - pub fn new(constraint: &str) -> Option { - constraint.parse().ok() - } - - /// Returns the category of this constraint violation. - pub fn categorize(&self) -> ConstraintCategory { - match self { - WorkspaceActivitiesConstraints::ParamsSize => ConstraintCategory::Validation, - } - } -} - -impl From for String { - #[inline] - fn from(val: WorkspaceActivitiesConstraints) -> Self { - val.to_string() - } -} - -impl TryFrom for WorkspaceActivitiesConstraints { - type Error = strum::ParseError; - - #[inline] - fn try_from(value: String) -> Result { - value.parse() - } -} diff --git a/crates/nvisy-postgres/src/types/constraint/workspace_connection_syncs.rs b/crates/nvisy-postgres/src/types/constraint/workspace_connection_syncs.rs index 9309043a..5914b670 100644 --- a/crates/nvisy-postgres/src/types/constraint/workspace_connection_syncs.rs +++ b/crates/nvisy-postgres/src/types/constraint/workspace_connection_syncs.rs @@ -1,71 +1,14 @@ //! Workspace connection syncs table constraint violations. -use serde::{Deserialize, Serialize}; -use strum::{Display, EnumIter, EnumString}; - -use super::ConstraintCategory; +use strum::EnumString; /// Workspace connection syncs table constraint violations. -#[derive(Debug, Clone, Copy, Eq, PartialEq)] -#[derive(Serialize, Deserialize, Display, EnumIter, EnumString)] -#[serde(into = "String", try_from = "String")] +#[derive(Debug, Clone, Copy, Eq, PartialEq, EnumString)] pub enum WorkspaceConnectionSyncConstraints { - // Size / validation constraints - #[strum(serialize = "workspace_connection_syncs_records_synced_non_negative")] - RecordsSyncedNonNegative, - #[strum(serialize = "workspace_connection_syncs_attempt_positive")] - AttemptPositive, #[strum(serialize = "workspace_connection_syncs_error_message_length")] ErrorMessageLength, #[strum(serialize = "workspace_connection_syncs_metadata_size")] MetadataSize, - - // Chronological constraints - #[strum(serialize = "workspace_connection_syncs_completed_after_started")] - CompletedAfterStarted, - - // Uniqueness constraints #[strum(serialize = "workspace_connection_syncs_one_active_idx")] OneActivePerConnection, } - -impl WorkspaceConnectionSyncConstraints { - /// Creates a new [`WorkspaceConnectionSyncConstraints`] from the constraint name. - pub fn new(constraint: &str) -> Option { - constraint.parse().ok() - } - - /// Returns the category of this constraint violation. - pub fn categorize(&self) -> ConstraintCategory { - match self { - WorkspaceConnectionSyncConstraints::RecordsSyncedNonNegative - | WorkspaceConnectionSyncConstraints::AttemptPositive - | WorkspaceConnectionSyncConstraints::ErrorMessageLength - | WorkspaceConnectionSyncConstraints::MetadataSize => ConstraintCategory::Validation, - - WorkspaceConnectionSyncConstraints::CompletedAfterStarted => { - ConstraintCategory::Chronological - } - - WorkspaceConnectionSyncConstraints::OneActivePerConnection => { - ConstraintCategory::Uniqueness - } - } - } -} - -impl From for String { - #[inline] - fn from(val: WorkspaceConnectionSyncConstraints) -> Self { - val.to_string() - } -} - -impl TryFrom for WorkspaceConnectionSyncConstraints { - type Error = strum::ParseError; - - #[inline] - fn try_from(value: String) -> Result { - value.parse() - } -} diff --git a/crates/nvisy-postgres/src/types/constraint/workspace_connections.rs b/crates/nvisy-postgres/src/types/constraint/workspace_connections.rs index 810f2c6e..da59c3b2 100644 --- a/crates/nvisy-postgres/src/types/constraint/workspace_connections.rs +++ b/crates/nvisy-postgres/src/types/constraint/workspace_connections.rs @@ -1,89 +1,24 @@ //! Workspace connections table constraint violations. -use serde::{Deserialize, Serialize}; -use strum::{Display, EnumIter, EnumString}; - -use super::ConstraintCategory; +use strum::EnumString; /// Workspace connections table constraint violations. -#[derive(Debug, Clone, Copy, Eq, PartialEq)] -#[derive(Serialize, Deserialize, Display, EnumIter, EnumString)] -#[serde(into = "String", try_from = "String")] +#[derive(Debug, Clone, Copy, Eq, PartialEq, EnumString)] pub enum WorkspaceConnectionConstraints { - // Name validation constraints #[strum(serialize = "workspace_connections_display_name_length")] NameLength, - - // Provider validation constraints #[strum(serialize = "workspace_connections_provider_length")] ProviderLength, - - // Data validation constraints #[strum(serialize = "workspace_connections_data_size")] DataSize, - - // Metadata validation constraints #[strum(serialize = "workspace_connections_metadata_size")] MetadataSize, - - // Schedule validation constraints #[strum(serialize = "workspace_connection_schedule_cron_length")] ScheduleCronLength, #[strum(serialize = "workspace_connection_schedule_import_only")] ScheduleImportOnly, - - // Uniqueness constraints #[strum(serialize = "workspace_connections_workspace_id_id_key")] WorkspaceIdIdUnique, #[strum(serialize = "workspace_connections_display_name_unique_idx")] NameUnique, - - // Chronological constraints - #[strum(serialize = "workspace_connections_updated_after_created")] - UpdatedAfterCreated, - #[strum(serialize = "workspace_connections_deleted_after_created")] - DeletedAfterCreated, -} - -impl WorkspaceConnectionConstraints { - /// Creates a new [`WorkspaceConnectionConstraints`] from the constraint name. - pub fn new(constraint: &str) -> Option { - constraint.parse().ok() - } - - /// Returns the category of this constraint violation. - pub fn categorize(&self) -> ConstraintCategory { - match self { - WorkspaceConnectionConstraints::NameLength - | WorkspaceConnectionConstraints::ProviderLength - | WorkspaceConnectionConstraints::DataSize - | WorkspaceConnectionConstraints::MetadataSize - | WorkspaceConnectionConstraints::ScheduleCronLength - | WorkspaceConnectionConstraints::ScheduleImportOnly => ConstraintCategory::Validation, - - WorkspaceConnectionConstraints::WorkspaceIdIdUnique - | WorkspaceConnectionConstraints::NameUnique => ConstraintCategory::Uniqueness, - - WorkspaceConnectionConstraints::UpdatedAfterCreated - | WorkspaceConnectionConstraints::DeletedAfterCreated => { - ConstraintCategory::Chronological - } - } - } -} - -impl From for String { - #[inline] - fn from(val: WorkspaceConnectionConstraints) -> Self { - val.to_string() - } -} - -impl TryFrom for WorkspaceConnectionConstraints { - type Error = strum::ParseError; - - #[inline] - fn try_from(value: String) -> Result { - value.parse() - } } diff --git a/crates/nvisy-postgres/src/types/constraint/workspace_invites.rs b/crates/nvisy-postgres/src/types/constraint/workspace_invites.rs index 45ef0390..faee1a3e 100644 --- a/crates/nvisy-postgres/src/types/constraint/workspace_invites.rs +++ b/crates/nvisy-postgres/src/types/constraint/workspace_invites.rs @@ -1,69 +1,12 @@ //! Workspace invites table constraint violations. -use serde::{Deserialize, Serialize}; -use strum::{Display, EnumIter, EnumString}; - -use super::ConstraintCategory; +use strum::EnumString; /// Workspace invites table constraint violations. -#[derive(Debug, Clone, Copy, Eq, PartialEq)] -#[derive(Serialize, Deserialize, Display, EnumIter, EnumString)] -#[serde(into = "String", try_from = "String")] +#[derive(Debug, Clone, Copy, Eq, PartialEq, EnumString)] pub enum WorkspaceInviteConstraints { - // Invite unique constraints #[strum(serialize = "workspace_invites_workspace_id_id_key")] WorkspaceIdIdUnique, - - // Invite validation constraints - #[strum(serialize = "workspace_invites_invite_token_not_empty")] - InviteTokenNotEmpty, #[strum(serialize = "workspace_invites_invitee_email_format")] InviteeEmailFormat, - - // Invite chronological constraints - #[strum(serialize = "workspace_invites_expires_after_created")] - ExpiresAfterCreated, - #[strum(serialize = "workspace_invites_updated_after_created")] - UpdatedAfterCreated, - #[strum(serialize = "workspace_invites_responded_after_created")] - RespondedAfterCreated, -} - -impl WorkspaceInviteConstraints { - /// Creates a new [`WorkspaceInviteConstraints`] from the constraint name. - pub fn new(constraint: &str) -> Option { - constraint.parse().ok() - } - - /// Returns the category of this constraint violation. - pub fn categorize(&self) -> ConstraintCategory { - match self { - WorkspaceInviteConstraints::WorkspaceIdIdUnique => ConstraintCategory::Uniqueness, - - WorkspaceInviteConstraints::InviteTokenNotEmpty - | WorkspaceInviteConstraints::InviteeEmailFormat => ConstraintCategory::Validation, - - WorkspaceInviteConstraints::ExpiresAfterCreated - | WorkspaceInviteConstraints::UpdatedAfterCreated - | WorkspaceInviteConstraints::RespondedAfterCreated => { - ConstraintCategory::Chronological - } - } - } -} - -impl From for String { - #[inline] - fn from(val: WorkspaceInviteConstraints) -> Self { - val.to_string() - } -} - -impl TryFrom for WorkspaceInviteConstraints { - type Error = strum::ParseError; - - #[inline] - fn try_from(value: String) -> Result { - value.parse() - } } diff --git a/crates/nvisy-postgres/src/types/constraint/workspace_members.rs b/crates/nvisy-postgres/src/types/constraint/workspace_members.rs index 3f126b18..d0bafb91 100644 --- a/crates/nvisy-postgres/src/types/constraint/workspace_members.rs +++ b/crates/nvisy-postgres/src/types/constraint/workspace_members.rs @@ -1,51 +1,10 @@ //! Workspace members table constraint violations. -use serde::{Deserialize, Serialize}; -use strum::{Display, EnumIter, EnumString}; - -use super::ConstraintCategory; +use strum::EnumString; /// Workspace members table constraint violations. -#[derive(Debug, Clone, Copy, Eq, PartialEq)] -#[derive(Serialize, Deserialize, Display, EnumIter, EnumString)] -#[serde(into = "String", try_from = "String")] +#[derive(Debug, Clone, Copy, Eq, PartialEq, EnumString)] pub enum WorkspaceMemberConstraints { - // Member uniqueness constraints #[strum(serialize = "workspace_members_pkey")] MembershipUnique, - - // Member chronological constraints - #[strum(serialize = "workspace_members_updated_after_created")] - UpdatedAfterCreated, -} - -impl WorkspaceMemberConstraints { - /// Creates a new [`WorkspaceMemberConstraints`] from the constraint name. - pub fn new(constraint: &str) -> Option { - constraint.parse().ok() - } - - /// Returns the category of this constraint violation. - pub fn categorize(&self) -> ConstraintCategory { - match self { - WorkspaceMemberConstraints::MembershipUnique => ConstraintCategory::Uniqueness, - WorkspaceMemberConstraints::UpdatedAfterCreated => ConstraintCategory::Chronological, - } - } -} - -impl From for String { - #[inline] - fn from(val: WorkspaceMemberConstraints) -> Self { - val.to_string() - } -} - -impl TryFrom for WorkspaceMemberConstraints { - type Error = strum::ParseError; - - #[inline] - fn try_from(value: String) -> Result { - value.parse() - } } diff --git a/crates/nvisy-postgres/src/types/constraint/workspace_policies.rs b/crates/nvisy-postgres/src/types/constraint/workspace_policies.rs index ff0205c2..5c613eda 100644 --- a/crates/nvisy-postgres/src/types/constraint/workspace_policies.rs +++ b/crates/nvisy-postgres/src/types/constraint/workspace_policies.rs @@ -1,16 +1,10 @@ //! Workspace policies table constraint violations. -use serde::{Deserialize, Serialize}; -use strum::{Display, EnumIter, EnumString}; - -use super::ConstraintCategory; +use strum::EnumString; /// Workspace policies table constraint violations. -#[derive(Debug, Clone, Copy, Eq, PartialEq)] -#[derive(Serialize, Deserialize, Display, EnumIter, EnumString)] -#[serde(into = "String", try_from = "String")] +#[derive(Debug, Clone, Copy, Eq, PartialEq, EnumString)] pub enum WorkspacePolicyConstraints { - // Validation constraints #[strum(serialize = "workspace_policies_slug_length")] SlugLength, #[strum(serialize = "workspace_policies_slug_format")] @@ -23,60 +17,10 @@ pub enum WorkspacePolicyConstraints { DefinitionSize, #[strum(serialize = "workspace_policies_metadata_size")] MetadataSize, - - // Uniqueness constraints #[strum(serialize = "workspace_policies_workspace_id_id_key")] WorkspaceIdIdUnique, #[strum(serialize = "workspace_policies_slug_unique_idx")] SlugUnique, #[strum(serialize = "workspace_policies_display_name_unique_idx")] NameUnique, - - // Chronological constraints - #[strum(serialize = "workspace_policies_updated_after_created")] - UpdatedAfterCreated, - #[strum(serialize = "workspace_policies_deleted_after_created")] - DeletedAfterCreated, -} - -impl WorkspacePolicyConstraints { - /// Creates a new [`WorkspacePolicyConstraints`] from the constraint name. - pub fn new(constraint: &str) -> Option { - constraint.parse().ok() - } - - /// Returns the category of this constraint violation. - pub fn categorize(&self) -> ConstraintCategory { - match self { - WorkspacePolicyConstraints::SlugLength - | WorkspacePolicyConstraints::SlugFormat - | WorkspacePolicyConstraints::NameLength - | WorkspacePolicyConstraints::DescriptionLength - | WorkspacePolicyConstraints::DefinitionSize - | WorkspacePolicyConstraints::MetadataSize => ConstraintCategory::Validation, - - WorkspacePolicyConstraints::WorkspaceIdIdUnique - | WorkspacePolicyConstraints::SlugUnique - | WorkspacePolicyConstraints::NameUnique => ConstraintCategory::Uniqueness, - - WorkspacePolicyConstraints::UpdatedAfterCreated - | WorkspacePolicyConstraints::DeletedAfterCreated => ConstraintCategory::Chronological, - } - } -} - -impl From for String { - #[inline] - fn from(val: WorkspacePolicyConstraints) -> Self { - val.to_string() - } -} - -impl TryFrom for WorkspacePolicyConstraints { - type Error = strum::ParseError; - - #[inline] - fn try_from(value: String) -> Result { - value.parse() - } } diff --git a/crates/nvisy-postgres/src/types/constraint/workspace_webhooks.rs b/crates/nvisy-postgres/src/types/constraint/workspace_webhooks.rs index b749b57d..abc2ee01 100644 --- a/crates/nvisy-postgres/src/types/constraint/workspace_webhooks.rs +++ b/crates/nvisy-postgres/src/types/constraint/workspace_webhooks.rs @@ -1,20 +1,12 @@ //! Workspace webhooks table constraint violations. -use serde::{Deserialize, Serialize}; -use strum::{Display, EnumIter, EnumString}; - -use super::ConstraintCategory; +use strum::EnumString; /// Workspace webhooks table constraint violations. -#[derive(Debug, Clone, Copy, Eq, PartialEq)] -#[derive(Serialize, Deserialize, Display, EnumIter, EnumString)] -#[serde(into = "String", try_from = "String")] +#[derive(Debug, Clone, Copy, Eq, PartialEq, EnumString)] pub enum WorkspaceWebhookConstraints { - // Webhook unique constraints #[strum(serialize = "workspace_webhooks_workspace_id_id_key")] WorkspaceIdIdUnique, - - // Webhook validation constraints #[strum(serialize = "workspace_webhooks_display_name_length")] DisplayNameLength, #[strum(serialize = "workspace_webhooks_description_length")] @@ -27,50 +19,4 @@ pub enum WorkspaceWebhookConstraints { EventsNotEmpty, #[strum(serialize = "workspace_webhooks_headers_size")] HeadersSize, - - // Webhook chronological constraints - #[strum(serialize = "workspace_webhooks_updated_after_created")] - UpdatedAfterCreated, - #[strum(serialize = "workspace_webhooks_deleted_after_created")] - DeletedAfterCreated, -} - -impl WorkspaceWebhookConstraints { - /// Creates a new [`WorkspaceWebhookConstraints`] from the constraint name. - pub fn new(constraint: &str) -> Option { - constraint.parse().ok() - } - - /// Returns the category of this constraint violation. - pub fn categorize(&self) -> ConstraintCategory { - match self { - WorkspaceWebhookConstraints::WorkspaceIdIdUnique => ConstraintCategory::Uniqueness, - - WorkspaceWebhookConstraints::DisplayNameLength - | WorkspaceWebhookConstraints::DescriptionLength - | WorkspaceWebhookConstraints::UrlLength - | WorkspaceWebhookConstraints::UrlFormat - | WorkspaceWebhookConstraints::EventsNotEmpty - | WorkspaceWebhookConstraints::HeadersSize => ConstraintCategory::Validation, - - WorkspaceWebhookConstraints::UpdatedAfterCreated - | WorkspaceWebhookConstraints::DeletedAfterCreated => ConstraintCategory::Chronological, - } - } -} - -impl From for String { - #[inline] - fn from(val: WorkspaceWebhookConstraints) -> Self { - val.to_string() - } -} - -impl TryFrom for WorkspaceWebhookConstraints { - type Error = strum::ParseError; - - #[inline] - fn try_from(value: String) -> Result { - value.parse() - } } diff --git a/crates/nvisy-postgres/src/types/constraint/workspaces.rs b/crates/nvisy-postgres/src/types/constraint/workspaces.rs index 51fe9450..07932bc1 100644 --- a/crates/nvisy-postgres/src/types/constraint/workspaces.rs +++ b/crates/nvisy-postgres/src/types/constraint/workspaces.rs @@ -1,16 +1,10 @@ //! Workspaces table constraint violations. -use serde::{Deserialize, Serialize}; -use strum::{Display, EnumIter, EnumString}; - -use super::ConstraintCategory; +use strum::EnumString; /// Workspace table constraint violations. -#[derive(Debug, Clone, Copy, Eq, PartialEq)] -#[derive(Serialize, Deserialize, Display, EnumIter, EnumString)] -#[serde(into = "String", try_from = "String")] +#[derive(Debug, Clone, Copy, Eq, PartialEq, EnumString)] pub enum WorkspaceConstraints { - // Workspace validation constraints #[strum(serialize = "workspaces_display_name_length")] DisplayNameLength, #[strum(serialize = "workspaces_slug_length")] @@ -23,61 +17,8 @@ pub enum WorkspaceConstraints { MetadataSize, #[strum(serialize = "workspaces_settings_size")] SettingsSize, - - // Workspace uniqueness constraints #[strum(serialize = "workspaces_slug_unique_idx")] SlugUnique, #[strum(serialize = "workspaces_display_name_owner_unique_idx")] NameUnique, - - // Workspace chronological constraints - #[strum(serialize = "workspaces_updated_after_created")] - UpdatedAfterCreated, - #[strum(serialize = "workspaces_deleted_after_created")] - DeletedAfterCreated, - #[strum(serialize = "workspaces_deleted_after_updated")] - DeletedAfterUpdated, -} - -impl WorkspaceConstraints { - /// Creates a new [`WorkspaceConstraints`] from the constraint name. - pub fn new(constraint: &str) -> Option { - constraint.parse().ok() - } - - /// Returns the category of this constraint violation. - pub fn categorize(&self) -> ConstraintCategory { - match self { - WorkspaceConstraints::DisplayNameLength - | WorkspaceConstraints::SlugLength - | WorkspaceConstraints::SlugFormat - | WorkspaceConstraints::DescriptionLengthMax - | WorkspaceConstraints::MetadataSize - | WorkspaceConstraints::SettingsSize => ConstraintCategory::Validation, - - WorkspaceConstraints::SlugUnique | WorkspaceConstraints::NameUnique => { - ConstraintCategory::Uniqueness - } - - WorkspaceConstraints::UpdatedAfterCreated - | WorkspaceConstraints::DeletedAfterCreated - | WorkspaceConstraints::DeletedAfterUpdated => ConstraintCategory::Chronological, - } - } -} - -impl From for String { - #[inline] - fn from(val: WorkspaceConstraints) -> Self { - val.to_string() - } -} - -impl TryFrom for WorkspaceConstraints { - type Error = strum::ParseError; - - #[inline] - fn try_from(value: String) -> Result { - value.parse() - } } diff --git a/crates/nvisy-postgres/src/types/enums/chat_role.rs b/crates/nvisy-postgres/src/types/enums/chat_role.rs new file mode 100644 index 00000000..f729807b --- /dev/null +++ b/crates/nvisy-postgres/src/types/enums/chat_role.rs @@ -0,0 +1,29 @@ +//! Chat message role enumeration. + +use diesel_derive_enum::DbEnum; +use serde::{Deserialize, Serialize}; +use strum::{Display, EnumIter, EnumString}; + +/// The author of a chat message. +/// +/// Corresponds to the `CHAT_ROLE` PostgreSQL enum. +#[derive(Debug, Clone, Copy, Eq, PartialEq, Hash)] +#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))] +#[derive(Serialize, Deserialize, DbEnum, Display, EnumIter, EnumString)] +#[ExistingTypePath = "crate::schema::sql_types::ChatRole"] +pub enum ChatRole { + /// A system instruction (server-authored context). + #[db_rename = "system"] + #[serde(rename = "system")] + System, + + /// A message from the account. + #[db_rename = "user"] + #[serde(rename = "user")] + User, + + /// A reply from the model. + #[db_rename = "assistant"] + #[serde(rename = "assistant")] + Assistant, +} diff --git a/crates/nvisy-postgres/src/types/enums/mod.rs b/crates/nvisy-postgres/src/types/enums/mod.rs index 2e00cf54..3ef202d1 100644 --- a/crates/nvisy-postgres/src/types/enums/mod.rs +++ b/crates/nvisy-postgres/src/types/enums/mod.rs @@ -8,6 +8,12 @@ pub mod api_token_type; pub mod notification_event; +// Chat-related enumerations +pub mod chat_role; + +// Connection-related enumerations +pub mod provider_type; + // Workspace-related enumerations pub mod activity_type; pub mod invite_status; @@ -29,12 +35,14 @@ pub mod pipeline_trigger_type; pub use activity_type::{ActivityCategory, ActivityType}; pub use api_token_type::ApiTokenType; +pub use chat_role::ChatRole; pub use file_kind::FileKind; pub use invite_status::InviteStatus; pub use notification_event::NotificationEvent; pub use pipeline_run_status::PipelineRunStatus; pub use pipeline_status::PipelineStatus; pub use pipeline_trigger_type::PipelineTriggerType; +pub use provider_type::ProviderType; pub use sync_deletion_policy::SyncDeletionPolicy; pub use sync_mode::SyncMode; pub use sync_status::SyncStatus; diff --git a/crates/nvisy-postgres/src/types/enums/provider_type.rs b/crates/nvisy-postgres/src/types/enums/provider_type.rs new file mode 100644 index 00000000..3c688a2b --- /dev/null +++ b/crates/nvisy-postgres/src/types/enums/provider_type.rs @@ -0,0 +1,28 @@ +//! Provider capability-category enumeration. + +use diesel_derive_enum::DbEnum; +use serde::{Deserialize, Serialize}; +use strum::{Display, EnumIter, EnumString}; + +/// The capability category of a connection's provider. +/// +/// Corresponds to the `PROVIDER_TYPE` PostgreSQL enum. A stable, closed set: +/// the concrete provider (the `provider` column, e.g. `s3` or `anthropic`) stays +/// open and extensible, while its capability is one of these types. Lets a +/// connection be found by what it can do — e.g. a workspace's language model — +/// without decrypting its config. +#[derive(Debug, Clone, Copy, Eq, PartialEq, Hash)] +#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))] +#[derive(Serialize, Deserialize, DbEnum, Display, EnumIter, EnumString)] +#[ExistingTypePath = "crate::schema::sql_types::ProviderType"] +pub enum ProviderType { + /// External object storage (s3, azure, gcs, ...). + #[db_rename = "object_store"] + #[serde(rename = "object_store")] + ObjectStore, + + /// LLM inference (openai, ollama, anthropic, ...). + #[db_rename = "language_model"] + #[serde(rename = "language_model")] + LanguageModel, +} diff --git a/crates/nvisy-postgres/src/types/mod.rs b/crates/nvisy-postgres/src/types/mod.rs index 7e5c88a9..a3b81b0e 100644 --- a/crates/nvisy-postgres/src/types/mod.rs +++ b/crates/nvisy-postgres/src/types/mod.rs @@ -14,16 +14,18 @@ mod utilities; pub use constants::{DEFAULT_RETENTION_DAYS, RECENTLY_SENT_HOURS}; pub use constraint::{ AccountApiTokenConstraints, AccountConstraints, AccountNotificationConstraints, - ConstraintCategory, ConstraintViolation, WorkspaceActivitiesConstraints, - WorkspaceConnectionConstraints, WorkspaceConnectionSyncConstraints, WorkspaceConstraints, - WorkspaceFileConstraints, WorkspaceInviteConstraints, WorkspaceMemberConstraints, - WorkspacePipelineConstraints, WorkspacePipelineReferenceConstraints, - WorkspacePipelineRunConstraints, WorkspacePolicyConstraints, WorkspaceWebhookConstraints, + ChatMessageConstraints, ChatSessionConstraints, ConstraintViolation, + WorkspaceActivitiesConstraints, WorkspaceConnectionConstraints, + WorkspaceConnectionSyncConstraints, WorkspaceConstraints, WorkspaceFileConstraints, + WorkspaceInviteConstraints, WorkspaceMemberConstraints, WorkspacePipelineConstraints, + WorkspacePipelineReferenceConstraints, WorkspacePipelineRunConstraints, + WorkspacePolicyConstraints, WorkspaceWebhookConstraints, }; pub use enums::{ - ActivityCategory, ActivityType, ApiTokenType, FileKind, InviteStatus, NotificationEvent, - PipelineRunStatus, PipelineStatus, PipelineTriggerType, SyncDeletionPolicy, SyncMode, - SyncStatus, SyncTriggerType, WebhookEvent, WebhookStatus, WorkspaceRole, + ActivityCategory, ActivityType, ApiTokenType, ChatRole, FileKind, InviteStatus, + NotificationEvent, PipelineRunStatus, PipelineStatus, PipelineTriggerType, ProviderType, + SyncDeletionPolicy, SyncMode, SyncStatus, SyncTriggerType, WebhookEvent, WebhookStatus, + WorkspaceRole, }; pub use filtering::{FileFilter, InviteFilter, MemberFilter, RunFilter}; pub use handle::{HANDLE_MAX_LENGTH, HANDLE_MIN_LENGTH, Handle, HandleError}; diff --git a/crates/nvisy-server/src/handler/chat.rs b/crates/nvisy-server/src/handler/chat.rs new file mode 100644 index 00000000..5f3b9bd6 --- /dev/null +++ b/crates/nvisy-server/src/handler/chat.rs @@ -0,0 +1,385 @@ +//! Assistant chat handlers: sessions and streaming messages. +//! +//! Chat is a workspace-scoped assistant. A session is a thread of messages; a +//! message POST persists the user's turn, streams the model's reply over SSE, +//! and persists the assembled reply when the stream ends. The model is the +//! workspace's language-model connection; it has no access to document contents. + +use aide::axum::ApiRouter; +use aide::transform::TransformOperation; +use async_stream::stream; +use axum::extract::State; +use axum::http::StatusCode; +use axum::response::sse::Event; +use futures::StreamExt; +use nvisy_postgres::PgClient; +use nvisy_postgres::model::NewChatSession; +use nvisy_postgres::query::{AppendSessionUpdate, ChatMessageRepository, ChatSessionRepository}; +use nvisy_postgres::types::ChatRole; +use tokio_util::sync::CancellationToken; + +use crate::extract::{ + AuthProvider, AuthState, Json, Path, Permission, Query, ValidateJson, WorkspaceContext, +}; +use crate::handler::request::{ + ChatSessionPathParams, CreateChatSession, OffsetPagination, SendChatMessage, +}; +use crate::handler::response::{ChatMessage, ChatSession, ChatSessionsPage, ErrorResponse}; +use crate::handler::utility::SseResponse; +use crate::handler::{Error, Result}; +use crate::service::{ChatService, ServiceState, TurnLocation}; + +/// Tracing target for chat operations. +const TRACING_TARGET: &str = "nvisy_server::handler::chat"; + +/// How long a session title seeded from the first message may be. +const TITLE_MAX: usize = 80; + +/// The default session title, until seeded from the first message. +const DEFAULT_TITLE: &str = "New chat"; + +/// Maximum assistant-reply length, in bytes of plaintext. Kept below the +/// encrypted-content column limit (131072 bytes) with headroom for the +/// encryption framing (nonce, tag, chunking), so an accepted reply always fits. +const MAX_REPLY_BYTES: usize = 96 * 1024; + +/// Creates a new chat session in the workspace. +#[tracing::instrument(skip_all, fields(account_id = %auth_state.account_id, workspace_id = %workspace.id))] +async fn create_session( + State(pg_client): State, + AuthState(auth_state): AuthState, + WorkspaceContext(workspace): WorkspaceContext, + ValidateJson(request): ValidateJson, +) -> Result<(StatusCode, Json)> { + let mut conn = pg_client.get_connection().await?; + auth_state + .authorize_workspace(&mut conn, workspace.id, Permission::ViewWorkspace) + .await?; + + let session = conn + .create_chat_session(NewChatSession { + workspace_id: workspace.id, + account_id: auth_state.account_id, + title: request.title.unwrap_or_else(|| DEFAULT_TITLE.to_owned()), + }) + .await?; + + tracing::info!(target: TRACING_TARGET, session_id = %session.id, "Chat session created"); + Ok((StatusCode::CREATED, Json(ChatSession::from_model(session)))) +} + +fn create_session_docs(op: TransformOperation) -> TransformOperation { + op.summary("Create chat session") + .description("Opens a new assistant chat session in the workspace.") + .response::<201, Json>() + .response::<401, Json>() + .response::<403, Json>() +} + +/// Lists the workspace's chat sessions, most recently active first. +#[tracing::instrument(skip_all, fields(account_id = %auth_state.account_id, workspace_id = %workspace.id))] +async fn list_sessions( + State(pg_client): State, + AuthState(auth_state): AuthState, + WorkspaceContext(workspace): WorkspaceContext, + Query(pagination): Query, +) -> Result<(StatusCode, Json)> { + let mut conn = pg_client.get_connection().await?; + auth_state + .authorize_workspace(&mut conn, workspace.id, Permission::ViewWorkspace) + .await?; + + let sessions = conn + .list_chat_sessions(workspace.id, pagination.into()) + .await?; + let items = sessions.into_iter().map(ChatSession::from_model).collect(); + + Ok(( + StatusCode::OK, + Json(ChatSessionsPage::new(items, None, None)), + )) +} + +fn list_sessions_docs(op: TransformOperation) -> TransformOperation { + op.summary("List chat sessions") + .description("Returns the workspace's chat sessions, most recently active first.") + .response::<200, Json>() + .response::<401, Json>() + .response::<403, Json>() +} + +/// Returns a session's messages in chronological order. +#[tracing::instrument(skip_all, fields(account_id = %auth_state.account_id, workspace_id = %workspace.id, session_id = %path_params.session_id))] +async fn list_messages( + State(pg_client): State, + State(chat): State, + AuthState(auth_state): AuthState, + WorkspaceContext(workspace): WorkspaceContext, + Path(path_params): Path, +) -> Result<(StatusCode, Json>)> { + let mut conn = pg_client.get_connection().await?; + auth_state + .authorize_workspace(&mut conn, workspace.id, Permission::ViewWorkspace) + .await?; + + // Scope the session to the workspace before reading its messages. + conn.find_chat_session_in_workspace(workspace.id, path_params.session_id) + .await? + .ok_or_else(|| Error::not_found("chat session"))?; + + let messages = conn.list_chat_messages(path_params.session_id).await?; + let items = messages + .into_iter() + .map(|message| ChatMessage::from_model(message, workspace.id, &chat)) + .collect::>>()?; + + Ok((StatusCode::OK, Json(items))) +} + +fn list_messages_docs(op: TransformOperation) -> TransformOperation { + op.summary("List chat messages") + .description("Returns a session's messages in chronological order.") + .response::<200, Json>>() + .response::<401, Json>() + .response::<403, Json>() + .response::<404, Json>() +} + +/// Deletes a chat session. +#[tracing::instrument(skip_all, fields(account_id = %auth_state.account_id, workspace_id = %workspace.id, session_id = %path_params.session_id))] +async fn delete_session( + State(pg_client): State, + AuthState(auth_state): AuthState, + WorkspaceContext(workspace): WorkspaceContext, + Path(path_params): Path, +) -> Result { + let mut conn = pg_client.get_connection().await?; + auth_state + .authorize_workspace(&mut conn, workspace.id, Permission::ViewWorkspace) + .await?; + + let deleted = conn + .delete_chat_session(workspace.id, path_params.session_id) + .await?; + if !deleted { + return Err(Error::not_found("chat session")); + } + + Ok(StatusCode::NO_CONTENT) +} + +fn delete_session_docs(op: TransformOperation) -> TransformOperation { + op.summary("Delete chat session") + .description("Soft-deletes a chat session.") + .response::<204, ()>() + .response::<401, Json>() + .response::<403, Json>() + .response::<404, Json>() +} + +/// Sends a message and streams the assistant's reply as Server-Sent Events. +/// +/// Persists the user message, streams the model's reply as `token` events, and +/// persists the assembled reply when the stream ends. Empty replies (the model +/// produced no text) are not persisted. +/// +/// Authenticated with a Bearer token; browsers should consume it via a `fetch` +/// stream rather than the native `EventSource`, which cannot send an +/// `Authorization` header. +#[tracing::instrument(skip_all, fields(account_id = %auth_state.account_id, workspace_id = %workspace.id, session_id = %path_params.session_id))] +async fn send_message( + State(pg_client): State, + State(chat): State, + State(shutdown): State, + AuthState(auth_state): AuthState, + WorkspaceContext(workspace): WorkspaceContext, + Path(path_params): Path, + ValidateJson(request): ValidateJson, +) -> Result> { + let session_id = path_params.session_id; + let workspace_id = workspace.id; + + let mut conn = pg_client.get_connection().await?; + auth_state + .authorize_workspace(&mut conn, workspace_id, Permission::ViewWorkspace) + .await?; + + // Scope the session to the workspace before writing to it. + let session = conn + .find_chat_session_in_workspace(workspace_id, session_id) + .await? + .ok_or_else(|| Error::not_found("chat session"))?; + + // An explicit parent must belong to this session. The composite FK enforces + // this at write time, but reject it here — before inference — for a clean 404 + // rather than a failed insert after the model has run. + if let Some(parent_id) = request.parent_id + && conn + .find_chat_message_in_session(session_id, parent_id) + .await? + .is_none() + { + return Err(Error::not_found("chat message")); + } + + // The turn extends the branch the client is on: an explicit parent, else the + // session's current leaf. + let user_turn = TurnLocation { + workspace_id, + session_id, + parent_id: request.parent_id.or(session.current_message_id), + }; + + // Open the model turn BEFORE persisting anything: resolving the workspace's + // language-model connection can fail (409 when none is configured), and a + // failed send must not leave an orphan user turn in the history. + let mut tokens = chat + .stream_turn(&mut conn, user_turn, &request.content) + .await?; + + // The turn resolved: persist the user message under the branch, advancing the + // active leaf to it and seeding the title on the first message — all in one + // transaction so the session state can't diverge from its messages. + let user_message = chat + .append_message( + &mut conn, + user_turn, + ChatRole::User, + &request.content, + AppendSessionUpdate { + advance_leaf: true, + title: (session.title == DEFAULT_TITLE).then(|| seeded_title(&request.content)), + }, + ) + .await?; + + // The assistant reply replies to the user message just stored. + let reply_turn = TurnLocation { + parent_id: Some(user_message.id), + ..user_turn + }; + + drop(conn); + + let stream = stream! { + let mut reply = String::new(); + // Only a normal end-of-stream (`None`) is a complete reply. A shutdown, a + // generation error, or exceeding the reply limit stops mid-reply; + // persisting that would store a partial turn as if the assistant had + // finished, corrupting later history. + let completed = loop { + tokio::select! { + // Server shutting down: end the open stream promptly so it does + // not block graceful shutdown. + () = shutdown.cancelled() => break false, + next = tokens.next() => match next { + Some(Ok(delta)) => { + // Cap the reply so it always fits the encrypted-content + // column: a longer reply would fail to persist after the + // user already saw it, silently dropping it from history. + if reply.len() + delta.len() > MAX_REPLY_BYTES { + tracing::warn!(target: TRACING_TARGET, "Chat reply exceeded the size limit; stopping"); + yield error_event("The response exceeded the maximum length and was stopped."); + break false; + } + reply.push_str(&delta); + yield token_event(&ChatToken { delta }); + } + // A generation error: surface it and stop. + Some(Err(err)) => { + tracing::warn!(target: TRACING_TARGET, error = %err, "Chat generation failed"); + yield error_event(&err.to_string()); + break false; + } + // Generation finished normally. + None => break true, + }, + } + }; + + // Persist the assembled reply only on normal completion (best-effort: the + // user already saw it), under the user message it answered. + if completed + && !reply.is_empty() + && let Err(err) = chat.persist_reply(reply_turn, &reply).await + { + tracing::error!(target: TRACING_TARGET, error = %err, "Failed to persist assistant reply"); + } + }; + + Ok(SseResponse::new(stream)) +} + +fn send_message_docs(op: TransformOperation) -> TransformOperation { + op.summary("Send chat message") + .description( + "Sends a message and streams the assistant's reply as Server-Sent \ + Events. Each event's `data` is a `ChatToken` delta. Authenticate \ + with a Bearer token via a `fetch`-based client; the native \ + `EventSource` cannot send an `Authorization` header. 409 when the \ + workspace has no language model connection configured.", + ) + .response::<401, Json>() + .response::<403, Json>() + .response::<404, Json>() + .response::<409, Json>() +} + +/// One streamed chunk of the assistant's reply. +#[derive( + Debug, + Clone, + serde::Serialize, + serde::Deserialize, + schemars::JsonSchema +)] +#[serde(rename_all = "camelCase")] +pub struct ChatToken { + /// The text delta. + pub delta: String, +} + +/// Builds a `token` SSE event carrying a reply delta. +fn token_event(token: &ChatToken) -> Event { + Event::default() + .event("token") + .json_data(token) + .unwrap_or_else(|_| Event::default().event("token")) +} + +/// Builds an `error` SSE event carrying a failure message. +fn error_event(message: &str) -> Event { + Event::default().event("error").data(message) +} + +/// A session title seeded from the first message: trimmed to a single line and +/// capped so it reads well in a session list. +fn seeded_title(content: &str) -> String { + let line = content.trim().lines().next().unwrap_or("").trim(); + let mut title: String = line.chars().take(TITLE_MAX).collect(); + if title.trim().is_empty() { + title = "New chat".to_owned(); + } + title +} + +/// Returns the chat routes. +pub fn routes() -> ApiRouter { + use aide::axum::routing::*; + + ApiRouter::new() + .api_route( + "/workspaces/{workspaceSlug}/chat/sessions/", + post_with(create_session, create_session_docs) + .get_with(list_sessions, list_sessions_docs), + ) + .api_route( + "/workspaces/{workspaceSlug}/chat/sessions/{sessionId}/", + delete_with(delete_session, delete_session_docs), + ) + .api_route( + "/workspaces/{workspaceSlug}/chat/sessions/{sessionId}/messages/", + get_with(list_messages, list_messages_docs).post_with(send_message, send_message_docs), + ) + .with_path_items(|item| item.tag("Chat")) +} diff --git a/crates/nvisy-server/src/handler/connections.rs b/crates/nvisy-server/src/handler/connections.rs index 921b1d03..58f63daf 100644 --- a/crates/nvisy-server/src/handler/connections.rs +++ b/crates/nvisy-server/src/handler/connections.rs @@ -88,9 +88,10 @@ async fn create_connection( validate_sync_input(sync)?; } - // The provider column is derived from the typed config so the two can never - // disagree; the full config is encrypted at rest. + // The provider and its capability type are derived from the typed config so + // they can never disagree with it; the full config is encrypted at rest. let provider = request.config.provider_id().to_owned(); + let provider_type = request.config.provider_type(); let encrypted_data = crypto.encrypt_json(workspace.id, &request.config)?; let new_connection = NewWorkspaceConnection { @@ -98,6 +99,7 @@ async fn create_connection( account_id: auth_state.account_id, display_name: request.display_name, provider, + provider_type, encrypted_data, is_active: request.is_active, metadata: None, diff --git a/crates/nvisy-server/src/handler/error/inference_error.rs b/crates/nvisy-server/src/handler/error/inference_error.rs new file mode 100644 index 00000000..90875840 --- /dev/null +++ b/crates/nvisy-server/src/handler/error/inference_error.rs @@ -0,0 +1,16 @@ +//! Inference error to HTTP error conversion. +//! +//! Maps `nvisy_inference::Error` onto an HTTP error. Building the provider +//! client or a completion failing at runtime is a server-side fault (the stored +//! connection is the operator's config, not the caller's input), so it surfaces +//! as an internal error with the provider's own message as context. + +use super::http_error::{Error as HttpError, ErrorKind}; + +impl<'a> From for HttpError<'a> { + fn from(error: nvisy_inference::Error) -> Self { + ErrorKind::InternalServerError + .with_message("Language model request failed") + .with_context(error.to_string()) + } +} diff --git a/crates/nvisy-server/src/handler/error/mod.rs b/crates/nvisy-server/src/handler/error/mod.rs index 51af8c57..12213c83 100644 --- a/crates/nvisy-server/src/handler/error/mod.rs +++ b/crates/nvisy-server/src/handler/error/mod.rs @@ -3,9 +3,11 @@ mod crypto_error; mod engine_error; mod http_error; +mod inference_error; mod nats_error; mod object_error; mod pg_account; +mod pg_chat; mod pg_document; mod pg_error; mod pg_pipeline; diff --git a/crates/nvisy-server/src/handler/error/pg_account.rs b/crates/nvisy-server/src/handler/error/pg_account.rs index c76d2830..bc375e32 100644 --- a/crates/nvisy-server/src/handler/error/pg_account.rs +++ b/crates/nvisy-server/src/handler/error/pg_account.rs @@ -42,12 +42,6 @@ impl From for Error<'static> { AccountConstraints::LocaleFormat => { ErrorKind::BadRequest.with_message("Invalid locale format") } - AccountConstraints::UpdatedAfterCreated - | AccountConstraints::DeletedAfterCreated - | AccountConstraints::DeletedAfterUpdated - | AccountConstraints::PasswordChangedAfterCreated => { - ErrorKind::InternalServerError.into_error() - } AccountConstraints::SuspendedNotAdmin => { ErrorKind::BadRequest.with_message("Admin accounts cannot be suspended") } @@ -66,11 +60,6 @@ impl From for Error<'static> { AccountApiTokenConstraints::NameLength => { ErrorKind::BadRequest.with_message("Token name is too long") } - AccountApiTokenConstraints::ExpiredAfterIssued - | AccountApiTokenConstraints::DeletedAfterIssued - | AccountApiTokenConstraints::LastUsedAfterIssued => { - ErrorKind::InternalServerError.into_error() - } }; error.with_resource("account_api_token") @@ -82,10 +71,6 @@ impl From for Error<'static> { let error = match constraint { AccountNotificationConstraints::ParamsSize => ErrorKind::BadRequest .with_message("Notification params must be between 2 and 4096 bytes"), - AccountNotificationConstraints::ExpiresAfterCreated => ErrorKind::BadRequest - .with_message("Notification expiration time must be after creation time"), - AccountNotificationConstraints::ReadAfterCreated => ErrorKind::BadRequest - .with_message("Notification read time must be after creation time"), }; error.with_resource("notification") diff --git a/crates/nvisy-server/src/handler/error/pg_chat.rs b/crates/nvisy-server/src/handler/error/pg_chat.rs new file mode 100644 index 00000000..9da07173 --- /dev/null +++ b/crates/nvisy-server/src/handler/error/pg_chat.rs @@ -0,0 +1,31 @@ +//! Chat-related constraint violation error handlers. + +use nvisy_postgres::types::{ChatMessageConstraints, ChatSessionConstraints}; + +use crate::handler::{Error, ErrorKind}; + +impl From for Error<'static> { + fn from(c: ChatSessionConstraints) -> Self { + let error = match c { + ChatSessionConstraints::TitleLength => ErrorKind::BadRequest + .with_message("Chat title must be between 1 and 255 characters"), + }; + error.with_resource("chat_session") + } +} + +impl From for Error<'static> { + fn from(c: ChatMessageConstraints) -> Self { + let error = match c { + ChatMessageConstraints::ContentSize => { + ErrorKind::BadRequest.with_message("Chat message is empty or too large") + } + // A parent must be in the same session; a caller supplying a + // cross-session parent is a bad request. + ChatMessageConstraints::IdSession | ChatMessageConstraints::Parent => { + ErrorKind::BadRequest.with_message("Parent message does not belong to this session") + } + }; + error.with_resource("chat_message") + } +} diff --git a/crates/nvisy-server/src/handler/error/pg_document.rs b/crates/nvisy-server/src/handler/error/pg_document.rs index 4e835cba..c5c94b4f 100644 --- a/crates/nvisy-server/src/handler/error/pg_document.rs +++ b/crates/nvisy-server/src/handler/error/pg_document.rs @@ -17,15 +17,6 @@ impl From for Error<'static> { WorkspaceFileConstraints::FileSizeMin => { ErrorKind::BadRequest.with_message("File size must be greater than or equal to 0") } - WorkspaceFileConstraints::StoragePathNotEmpty => { - ErrorKind::InternalServerError.into_error() - } - WorkspaceFileConstraints::StorageBucketNotEmpty => { - ErrorKind::InternalServerError.into_error() - } - WorkspaceFileConstraints::FileHashSha256Length => { - ErrorKind::InternalServerError.into_error() - } WorkspaceFileConstraints::MetadataSize => { ErrorKind::BadRequest.with_message("File metadata size is invalid") } @@ -38,12 +29,6 @@ impl From for Error<'static> { WorkspaceFileConstraints::WorkspaceIdIdUnique => { ErrorKind::Conflict.with_message("A file with this identifier already exists") } - WorkspaceFileConstraints::UpdatedAfterCreated - | WorkspaceFileConstraints::DeletedAfterCreated - | WorkspaceFileConstraints::DeletedAfterUpdated - | WorkspaceFileConstraints::ExpiresAfterCreated => { - ErrorKind::InternalServerError.into_error() - } }; error.with_resource("file") diff --git a/crates/nvisy-server/src/handler/error/pg_error.rs b/crates/nvisy-server/src/handler/error/pg_error.rs index c1c830c4..62958722 100644 --- a/crates/nvisy-server/src/handler/error/pg_error.rs +++ b/crates/nvisy-server/src/handler/error/pg_error.rs @@ -20,6 +20,8 @@ impl From for Error<'static> { ConstraintViolation::Account(c) => c.into(), ConstraintViolation::AccountNotification(c) => c.into(), ConstraintViolation::AccountApiToken(c) => c.into(), + ConstraintViolation::ChatSession(c) => c.into(), + ConstraintViolation::ChatMessage(c) => c.into(), ConstraintViolation::Workspace(c) => c.into(), ConstraintViolation::WorkspaceMember(c) => c.into(), ConstraintViolation::WorkspaceInvite(c) => c.into(), diff --git a/crates/nvisy-server/src/handler/error/pg_pipeline.rs b/crates/nvisy-server/src/handler/error/pg_pipeline.rs index 8b201ce7..84f3ede2 100644 --- a/crates/nvisy-server/src/handler/error/pg_pipeline.rs +++ b/crates/nvisy-server/src/handler/error/pg_pipeline.rs @@ -37,10 +37,6 @@ impl From for Error<'static> { } WorkspacePipelineConstraints::WorkspaceIdIdUnique => ErrorKind::Conflict .with_message("A pipeline with this identifier already exists"), - WorkspacePipelineConstraints::UpdatedAfterCreated - | WorkspacePipelineConstraints::DeletedAfterCreated => { - ErrorKind::InternalServerError.into_error() - } }; error.with_resource("pipeline") @@ -57,9 +53,6 @@ impl From for Error<'static> { .with_message("Idempotency key must be 1 to 255 characters"), WorkspacePipelineRunConstraints::IdempotencyUnique => ErrorKind::Conflict .with_message("A run with this idempotency key already exists"), - WorkspacePipelineRunConstraints::CompletedAfterStarted => { - ErrorKind::InternalServerError.into_error() - } }; error.with_resource("pipeline_run") @@ -68,20 +61,12 @@ impl From for Error<'static> { impl From for Error<'static> { fn from(c: WorkspacePipelineReferenceConstraints) -> Self { - let (resource, error) = match c { - WorkspacePipelineReferenceConstraints::PolicyReference => ( - "policy", - ErrorKind::BadRequest - .with_message("Referenced policy does not exist in this workspace"), - ), - // The pipeline side of the FK only fails if the pipeline row vanished - // mid-transaction, which is a server-side fault rather than bad input. - WorkspacePipelineReferenceConstraints::PolicyPipelineReference => { - ("pipeline", ErrorKind::InternalServerError.into_error()) - } + let error = match c { + WorkspacePipelineReferenceConstraints::PolicyReference => ErrorKind::BadRequest + .with_message("Referenced policy does not exist in this workspace"), }; - error.with_resource(resource) + error.with_resource("policy") } } @@ -109,10 +94,6 @@ impl From for Error<'static> { WorkspaceConnectionConstraints::NameUnique => { ErrorKind::Conflict.with_message("A connection with this name already exists") } - WorkspaceConnectionConstraints::UpdatedAfterCreated - | WorkspaceConnectionConstraints::DeletedAfterCreated => { - ErrorKind::InternalServerError.into_error() - } }; error.with_resource("workspace_connection") @@ -130,11 +111,6 @@ impl From for Error<'static> { WorkspaceConnectionSyncConstraints::OneActivePerConnection => { ErrorKind::Conflict.with_message("A sync is already in progress") } - WorkspaceConnectionSyncConstraints::RecordsSyncedNonNegative - | WorkspaceConnectionSyncConstraints::AttemptPositive - | WorkspaceConnectionSyncConstraints::CompletedAfterStarted => { - ErrorKind::InternalServerError.into_error() - } }; error.with_resource("workspace_connection_sync") @@ -168,10 +144,6 @@ impl From for Error<'static> { WorkspacePolicyConstraints::WorkspaceIdIdUnique => { ErrorKind::Conflict.with_message("A policy with this identifier already exists") } - WorkspacePolicyConstraints::UpdatedAfterCreated - | WorkspacePolicyConstraints::DeletedAfterCreated => { - ErrorKind::InternalServerError.into_error() - } }; error.with_resource("workspace_policy") diff --git a/crates/nvisy-server/src/handler/error/pg_workspace.rs b/crates/nvisy-server/src/handler/error/pg_workspace.rs index e796016e..9e0609b6 100644 --- a/crates/nvisy-server/src/handler/error/pg_workspace.rs +++ b/crates/nvisy-server/src/handler/error/pg_workspace.rs @@ -32,11 +32,6 @@ impl From for Error<'static> { WorkspaceConstraints::SettingsSize => { ErrorKind::BadRequest.with_message("Workspace settings size is invalid") } - WorkspaceConstraints::UpdatedAfterCreated - | WorkspaceConstraints::DeletedAfterCreated - | WorkspaceConstraints::DeletedAfterUpdated => { - ErrorKind::InternalServerError.into_error() - } }; error.with_resource("workspace") @@ -48,9 +43,6 @@ impl From for Error<'static> { let error = match c { WorkspaceMemberConstraints::MembershipUnique => ErrorKind::Conflict .with_message("This account is already a member of the workspace"), - WorkspaceMemberConstraints::UpdatedAfterCreated => { - ErrorKind::InternalServerError.into_error() - } }; error.with_resource("workspace_member") @@ -66,14 +58,6 @@ impl From for Error<'static> { WorkspaceInviteConstraints::WorkspaceIdIdUnique => { ErrorKind::Conflict.with_message("An invite with this identifier already exists") } - WorkspaceInviteConstraints::InviteTokenNotEmpty => { - ErrorKind::InternalServerError.into_error() - } - WorkspaceInviteConstraints::ExpiresAfterCreated - | WorkspaceInviteConstraints::UpdatedAfterCreated - | WorkspaceInviteConstraints::RespondedAfterCreated => { - ErrorKind::InternalServerError.into_error() - } }; error.with_resource("workspace_invite") @@ -115,10 +99,6 @@ impl From for Error<'static> { WorkspaceWebhookConstraints::WorkspaceIdIdUnique => { ErrorKind::Conflict.with_message("A webhook with this identifier already exists") } - WorkspaceWebhookConstraints::UpdatedAfterCreated - | WorkspaceWebhookConstraints::DeletedAfterCreated => { - ErrorKind::InternalServerError.into_error() - } }; error.with_resource("workspace_webhook") diff --git a/crates/nvisy-server/src/handler/mod.rs b/crates/nvisy-server/src/handler/mod.rs index b8189ff7..0c0a3e2f 100644 --- a/crates/nvisy-server/src/handler/mod.rs +++ b/crates/nvisy-server/src/handler/mod.rs @@ -7,6 +7,7 @@ mod accounts; mod authentication; mod avatars; mod catalog; +mod chat; mod connection_syncs; mod connections; mod error; @@ -80,6 +81,7 @@ fn private_routes( .merge(workspaces::routes()) .merge(members::routes()) .merge(connections::routes()) + .merge(chat::routes()) .merge(connection_syncs::routes()) .merge(files::routes()) .merge(pipelines::routes()) diff --git a/crates/nvisy-server/src/handler/request/chat.rs b/crates/nvisy-server/src/handler/request/chat.rs new file mode 100644 index 00000000..5a0a093b --- /dev/null +++ b/crates/nvisy-server/src/handler/request/chat.rs @@ -0,0 +1,38 @@ +//! Assistant chat request types. + +use schemars::JsonSchema; +use serde::{Deserialize, Serialize}; +use uuid::Uuid; +use validator::Validate; + +/// Path parameters for a chat session. +#[must_use] +#[derive(Debug, Serialize, Deserialize, JsonSchema)] +#[serde(rename_all = "camelCase")] +pub struct ChatSessionPathParams { + /// The session id. + pub session_id: Uuid, +} + +/// Request to create a chat session. +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema, Validate)] +#[serde(rename_all = "camelCase")] +pub struct CreateChatSession { + /// Optional title. Defaults to a title seeded from the first message. + #[validate(length(min = 1, max = 255))] + pub title: Option, +} + +/// Request to send a message and stream the assistant's reply. +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema, Validate)] +#[serde(rename_all = "camelCase")] +pub struct SendChatMessage { + /// The user's message. + #[validate(length(min = 1, max = 65536))] + pub content: String, + /// The message this turn replies to (the branch being extended). Omit to + /// continue from the session's current leaf; use an earlier message's id to + /// branch (e.g. edit-and-resend). + #[serde(default)] + pub parent_id: Option, +} diff --git a/crates/nvisy-server/src/handler/request/mod.rs b/crates/nvisy-server/src/handler/request/mod.rs index 22009a69..50a5331e 100644 --- a/crates/nvisy-server/src/handler/request/mod.rs +++ b/crates/nvisy-server/src/handler/request/mod.rs @@ -2,6 +2,7 @@ mod accounts; mod authentications; +mod chat; mod connection_syncs; mod connections; mod files; @@ -19,6 +20,7 @@ mod workspaces; pub use accounts::*; pub use authentications::*; +pub use chat::*; pub use connection_syncs::*; pub use connections::*; pub use files::*; diff --git a/crates/nvisy-server/src/handler/response/chat.rs b/crates/nvisy-server/src/handler/response/chat.rs new file mode 100644 index 00000000..d7542950 --- /dev/null +++ b/crates/nvisy-server/src/handler/response/chat.rs @@ -0,0 +1,81 @@ +//! Assistant chat response types. + +use jiff::Timestamp; +use nvisy_postgres::model::{ChatMessage as ChatMessageModel, ChatSession as ChatSessionModel}; +use nvisy_postgres::types::ChatRole; +use schemars::JsonSchema; +use serde::{Deserialize, Serialize}; +use uuid::Uuid; + +use super::Page; +use crate::handler::Result; +use crate::service::ChatService; + +/// A chat session. +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +#[serde(rename_all = "camelCase")] +pub struct ChatSession { + /// Unique session identifier. + pub id: Uuid, + /// Human-readable title. + pub title: String, + /// Active leaf of the message tree (the conversation's resume point). + #[serde(skip_serializing_if = "Option::is_none")] + pub current_message_id: Option, + /// When the session was created. + pub created_at: Timestamp, + /// When the session was last active. + pub updated_at: Timestamp, +} + +impl ChatSession { + /// Builds the response from a stored session. + pub fn from_model(session: ChatSessionModel) -> Self { + Self { + id: session.id, + title: session.title, + current_message_id: session.current_message_id, + created_at: session.created_at.into(), + updated_at: session.updated_at.into(), + } + } +} + +/// Paginated list of chat sessions. +pub type ChatSessionsPage = Page; + +/// A single chat message. +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +#[serde(rename_all = "camelCase")] +pub struct ChatMessage { + /// Unique message identifier. + pub id: Uuid, + /// Parent in the conversation tree; absent for a root. + #[serde(skip_serializing_if = "Option::is_none")] + pub parent_id: Option, + /// Author of the message. + pub role: ChatRole, + /// Message text. + pub content: String, + /// When the message was created. + pub created_at: Timestamp, +} + +impl ChatMessage { + /// Builds the response from a stored message, decrypting its content under + /// the workspace key. + pub fn from_model( + message: ChatMessageModel, + workspace_id: Uuid, + chat: &ChatService, + ) -> Result { + let content = chat.decrypt_content(workspace_id, &message)?; + Ok(Self { + id: message.id, + parent_id: message.parent_id, + role: message.role, + content, + created_at: message.created_at.into(), + }) + } +} diff --git a/crates/nvisy-server/src/handler/response/connections.rs b/crates/nvisy-server/src/handler/response/connections.rs index 8b701ea3..68e2c3e1 100644 --- a/crates/nvisy-server/src/handler/response/connections.rs +++ b/crates/nvisy-server/src/handler/response/connections.rs @@ -2,7 +2,7 @@ use jiff::Timestamp; use nvisy_postgres::model::{WorkspaceConnection, WorkspaceConnectionSchedule}; -use nvisy_postgres::types::{ConnectionId, Handle, SyncDeletionPolicy, SyncMode}; +use nvisy_postgres::types::{ConnectionId, Handle, ProviderType, SyncDeletionPolicy, SyncMode}; use schemars::JsonSchema; use serde::{Deserialize, Serialize}; @@ -41,6 +41,8 @@ pub struct Connection { pub display_name: String, /// Provider identifier (`s3`, `azure`, `gcs`, `openai`, `ollama`, ...). pub provider: String, + /// Capability category of the provider (object store, language model, ...). + pub provider_type: ProviderType, /// Whether the connection is enabled. pub is_active: bool, /// Sync configuration; present only for sync-capable connections. @@ -105,6 +107,7 @@ impl Connection { created_by, display_name: connection.display_name, provider: connection.provider, + provider_type: connection.provider_type, is_active: connection.is_active, sync, created_at: connection.created_at.into(), diff --git a/crates/nvisy-server/src/handler/response/mod.rs b/crates/nvisy-server/src/handler/response/mod.rs index 9ec60744..5bdf4fb3 100644 --- a/crates/nvisy-server/src/handler/response/mod.rs +++ b/crates/nvisy-server/src/handler/response/mod.rs @@ -9,6 +9,7 @@ mod accounts; mod activities; mod authentications; mod catalog; +mod chat; mod connection_syncs; mod connections; mod errors; @@ -29,6 +30,7 @@ pub use accounts::*; pub use activities::*; pub use authentications::*; pub use catalog::*; +pub use chat::*; pub use connection_syncs::*; pub use connections::*; pub use errors::*; diff --git a/crates/nvisy-server/src/service/chat.rs b/crates/nvisy-server/src/service/chat.rs new file mode 100644 index 00000000..c3efe51b --- /dev/null +++ b/crates/nvisy-server/src/service/chat.rs @@ -0,0 +1,190 @@ +//! Assistant chat service. +//! +//! [`ChatService`] resolves a workspace's language-model connection into an +//! [`InferenceClient`], persists a session's messages (encrypting their content +//! under the workspace key), and drives a streaming chat turn against the +//! session's history. + +use nvisy_inference::{ChatTurn, InferenceClient, TokenStream}; +use nvisy_postgres::PgConn; +use nvisy_postgres::model::{ChatMessage, NewChatMessage}; +use nvisy_postgres::query::{ + AppendSessionUpdate, ChatMessageRepository, WorkspaceConnectionRepository, +}; +use nvisy_postgres::types::{ChatRole, ProviderType}; +use uuid::Uuid; + +use crate::handler::{ErrorKind, Result}; +use crate::service::{ConnectionConfig, Infra}; + +/// Where in a conversation a turn happens: the workspace and session it belongs +/// to, and the message it extends (its parent in the tree; `None` starts a new +/// root). +#[derive(Debug, Clone, Copy)] +pub struct TurnLocation { + /// Workspace owning the session (and its encryption key + model connection). + pub workspace_id: Uuid, + /// Session the turn belongs to. + pub session_id: Uuid, + /// The message this turn replies to; `None` is a root. + pub parent_id: Option, +} + +/// The assistant's system preamble. Kept deliberately narrow: this is a plain +/// chat assistant with no access to document contents (a hard constraint on a +/// redaction platform). +const PREAMBLE: &str = "You are the assistant for a document redaction platform. \ + Help the user understand and operate their workspace: redaction policies, \ + detections, and pipelines. You do not have access to the contents of any \ + document. Be concise and accurate."; + +/// Resolves a workspace's inference backend and drives streaming chat turns. +/// +/// Cloneable and cheap to pass around: holds the shared [`Infra`] clients (all +/// `Arc`-backed) and takes the per-request database connection as a method +/// argument. +#[derive(Clone)] +#[must_use = "service does nothing unless you use it"] +pub struct ChatService { + infra: Infra, +} + +impl ChatService { + /// Creates a new [`ChatService`]. + pub fn new(infra: Infra) -> Self { + Self { infra } + } + + /// Resolves the workspace's language-model connection into an inference + /// client. + /// + /// Errors when the workspace has no language-model connection configured + /// (`409 Conflict`), or when its stored config is not an inference config or + /// cannot build a client (`500`). + async fn resolve_client( + &self, + conn: &mut PgConn, + workspace_id: Uuid, + ) -> Result { + let connection = conn + .find_connection_by_type(workspace_id, ProviderType::LanguageModel) + .await? + .ok_or_else(|| { + ErrorKind::Conflict + .with_message("This workspace has no language model connection configured") + .with_resource("connection") + })?; + + let config: ConnectionConfig = self + .infra + .crypto + .decrypt_json(workspace_id, &connection.encrypted_data)?; + + let ConnectionConfig::Inference(llm) = config else { + return Err(ErrorKind::InternalServerError + .with_message("Connection is not a language model connection")); + }; + + llm.connect(None).map_err(|err| { + ErrorKind::InternalServerError + .with_message("Failed to build the language model client") + .with_context(err.to_string()) + }) + } + + /// Streams the assistant's reply to `prompt`, using the conversation path + /// ending at `parent_id` as context. + /// + /// Loads the session's messages, walks the path (root → `parent_id`), + /// decrypts it, and streams the model's reply. Resolving the model connection + /// happens first, so a missing connection fails before the caller persists + /// anything. Returns a [`TokenStream`] of text deltas. + pub async fn stream_turn( + &self, + conn: &mut PgConn, + at: TurnLocation, + prompt: &str, + ) -> Result { + let client = self.resolve_client(conn, at.workspace_id).await?; + let messages = conn.list_chat_messages(at.session_id).await?; + let path = ChatMessage::path_to(&messages, at.parent_id); + let history = self.to_history(at.workspace_id, &path)?; + Ok(client.stream_chat(prompt, history)) + } + + /// Appends a message at `at` in the session's tree — encrypting its content + /// under the workspace key — and applies `session_update` (advance the active + /// leaf, set the title) in the same transaction, so a message and the session + /// state it implies never diverge. Returns the stored row. + pub async fn append_message( + &self, + conn: &mut PgConn, + at: TurnLocation, + role: ChatRole, + text: &str, + session_update: AppendSessionUpdate, + ) -> Result { + let content = self + .infra + .crypto + .encrypt(at.workspace_id, text.as_bytes())?; + Ok(conn + .append_chat_message( + NewChatMessage { + session_id: at.session_id, + parent_id: at.parent_id, + role, + content, + }, + session_update, + ) + .await?) + } + + /// Persists the assistant's assembled reply at `at`, advancing the session's + /// active leaf to it in the same transaction. + /// + /// Acquires its own connection: it runs after the stream completes, when the + /// request connection has already been released back to the pool. + pub async fn persist_reply(&self, at: TurnLocation, reply: &str) -> Result<()> { + let mut conn = self.infra.postgres.get_connection().await?; + self.append_message( + &mut conn, + at, + ChatRole::Assistant, + reply, + AppendSessionUpdate { + advance_leaf: true, + ..Default::default() + }, + ) + .await?; + Ok(()) + } + + /// Decrypts a stored message's content under the workspace key. + pub fn decrypt_content(&self, workspace_id: Uuid, message: &ChatMessage) -> Result { + let bytes = self.infra.crypto.decrypt(workspace_id, &message.content)?; + String::from_utf8(bytes).map_err(|err| { + ErrorKind::InternalServerError + .with_message("Stored chat message is not valid UTF-8") + .with_context(err.to_string()) + }) + } + + /// Builds the chat history from a decrypted path, preceded by the assistant + /// preamble as a system instruction. + fn to_history(&self, workspace_id: Uuid, path: &[&ChatMessage]) -> Result> { + let mut history = Vec::with_capacity(path.len() + 1); + history.push(ChatTurn::system(PREAMBLE)); + for message in path { + let content = self.decrypt_content(workspace_id, message)?; + history.push(match message.role { + ChatRole::System => ChatTurn::system(content), + ChatRole::User => ChatTurn::user(content), + ChatRole::Assistant => ChatTurn::assistant(content), + }); + } + Ok(history) + } +} diff --git a/crates/nvisy-server/src/service/connection_config.rs b/crates/nvisy-server/src/service/connection_config.rs index 5d8f97c7..abe59c67 100644 --- a/crates/nvisy-server/src/service/connection_config.rs +++ b/crates/nvisy-server/src/service/connection_config.rs @@ -8,6 +8,7 @@ use nvisy_inference::providers::LlmConfig; use nvisy_object::providers::StorageConfig; +use nvisy_postgres::types::ProviderType; use schemars::JsonSchema; use serde::{Deserialize, Serialize}; @@ -35,6 +36,16 @@ impl ConnectionConfig { } } + /// The capability category of this config, stored on the connection so it can + /// be found by what it does without decrypting the config. + #[must_use] + pub fn provider_type(&self) -> ProviderType { + match self { + Self::ObjectStore(_) => ProviderType::ObjectStore, + Self::Inference(_) => ProviderType::LanguageModel, + } + } + /// Whether this connection has the sync capability (object stores do; LLM /// connections do not). Determines whether sync configuration and syncs /// apply. diff --git a/crates/nvisy-server/src/service/mod.rs b/crates/nvisy-server/src/service/mod.rs index 94e94a30..1ecbf7c5 100644 --- a/crates/nvisy-server/src/service/mod.rs +++ b/crates/nvisy-server/src/service/mod.rs @@ -1,6 +1,7 @@ //! Application state and dependency injection. mod avatar; +mod chat; mod connection_config; mod crypto; mod detection; @@ -27,6 +28,7 @@ use nvisy_webhook::WebhookService; use tokio_util::sync::CancellationToken; pub use crate::service::avatar::{AVATAR_CONTENT_TYPE, AvatarService, MAX_AVATAR_UPLOAD_BYTES}; +pub use crate::service::chat::{ChatService, TurnLocation}; pub use crate::service::connection_config::ConnectionConfig; pub use crate::service::crypto::{CryptoConfig, CryptoService}; pub(crate) use crate::service::crypto::{CryptoError, HashingReader, Measurements}; @@ -267,6 +269,7 @@ impl_di_field!( // Stateless services, composed from `Infra` on extraction: impl_di_compose!( AvatarService => AvatarService::new, + ChatService => ChatService::new, RunBlobStore => RunBlobStore::new, DetectionQueue => DetectionQueue::new, WebhookEmitter => WebhookEmitter::new, diff --git a/migrations/2026-01-19-045013_connections/down.sql b/migrations/2026-01-19-045013_connections/down.sql index fb327c25..faea51c5 100644 --- a/migrations/2026-01-19-045013_connections/down.sql +++ b/migrations/2026-01-19-045013_connections/down.sql @@ -8,6 +8,8 @@ DROP TABLE IF EXISTS workspace_connection_schedule; DROP TABLE IF EXISTS workspace_connections; +DROP TYPE IF EXISTS PROVIDER_TYPE; + DROP TYPE IF EXISTS SYNC_DELETION_POLICY; DROP TYPE IF EXISTS SYNC_MODE; diff --git a/migrations/2026-01-19-045013_connections/up.sql b/migrations/2026-01-19-045013_connections/up.sql index c31baed5..d06caea5 100644 --- a/migrations/2026-01-19-045013_connections/up.sql +++ b/migrations/2026-01-19-045013_connections/up.sql @@ -47,6 +47,18 @@ CREATE TYPE SYNC_DELETION_POLICY AS ENUM ( COMMENT ON TYPE SYNC_DELETION_POLICY IS 'How an import reconciles files whose source object has been deleted.'; +-- The capability category of a connection's provider. Stable, closed set: the +-- concrete provider (`provider` column) stays open and extensible, but its +-- capability is one of these types. Lets a connection be found by what it can do +-- (e.g. the workspace's language model) without decrypting its config. +CREATE TYPE PROVIDER_TYPE AS ENUM ( + 'object_store', -- External object storage (s3, azure, gcs) + 'language_model' -- LLM inference (openai, ollama, anthropic) +); + +COMMENT ON TYPE PROVIDER_TYPE IS + 'Capability category of a connection provider (object store, language model, ...).'; + -- Workspace connections table (generic encrypted provider credentials) CREATE TABLE workspace_connections ( -- Primary identifier @@ -61,7 +73,11 @@ CREATE TABLE workspace_connections ( -- Core attributes display_name TEXT NOT NULL, + -- The concrete provider (open, extensible: 's3', 'anthropic', ...) and its + -- capability category (a stable, closed enum). The category lets a + -- connection be found by what it can do without decrypting its config. provider TEXT NOT NULL, + provider_type PROVIDER_TYPE NOT NULL, CONSTRAINT workspace_connections_display_name_length CHECK (length(trim(display_name)) BETWEEN 1 AND 255), CONSTRAINT workspace_connections_provider_length CHECK (length(trim(provider)) BETWEEN 1 AND 64), @@ -102,6 +118,12 @@ CREATE INDEX workspace_connections_provider_idx ON workspace_connections (provider, workspace_id) WHERE deleted_at IS NULL; +-- Find a workspace's connection of a given capability (e.g. its language model), +-- most recently updated first. +CREATE INDEX workspace_connections_provider_type_idx + ON workspace_connections (workspace_id, provider_type, updated_at DESC) + WHERE deleted_at IS NULL; + CREATE UNIQUE INDEX workspace_connections_display_name_unique_idx ON workspace_connections (workspace_id, lower(trim(display_name))) WHERE deleted_at IS NULL; @@ -119,6 +141,7 @@ COMMENT ON COLUMN workspace_connections.workspace_id IS 'Parent workspace refere COMMENT ON COLUMN workspace_connections.account_id IS 'Creator account reference'; COMMENT ON COLUMN workspace_connections.display_name IS 'Human-readable connection display name (1-255 chars)'; COMMENT ON COLUMN workspace_connections.provider IS 'Provider identifier (e.g. s3, azure, gcs, openai, ollama, anthropic)'; +COMMENT ON COLUMN workspace_connections.provider_type IS 'Capability category of the provider (object_store, language_model)'; COMMENT ON COLUMN workspace_connections.encrypted_data IS 'XChaCha20-Poly1305 encrypted JSON: provider config + credentials'; COMMENT ON COLUMN workspace_connections.is_active IS 'Whether the connection is enabled'; COMMENT ON COLUMN workspace_connections.metadata IS 'Non-encrypted metadata for filtering/display'; diff --git a/migrations/2026-08-19-034709_chat/down.sql b/migrations/2026-08-19-034709_chat/down.sql new file mode 100644 index 00000000..ea0b8a51 --- /dev/null +++ b/migrations/2026-08-19-034709_chat/down.sql @@ -0,0 +1,5 @@ +-- Revert the chat feature: drop messages, sessions, then the role enum. + +DROP TABLE IF EXISTS chat_messages; +DROP TABLE IF EXISTS chat_sessions; +DROP TYPE IF EXISTS CHAT_ROLE; diff --git a/migrations/2026-08-19-034709_chat/up.sql b/migrations/2026-08-19-034709_chat/up.sql new file mode 100644 index 00000000..d95f1cb8 --- /dev/null +++ b/migrations/2026-08-19-034709_chat/up.sql @@ -0,0 +1,114 @@ +-- Chat: workspace-scoped assistant conversations. A standalone workspace +-- resource. Each session is a thread of messages; the assistant's replies are +-- produced by the workspace's inference connection. + +-- Role of a chat message: who authored it. +CREATE TYPE CHAT_ROLE AS ENUM ( + 'system', -- System instruction (server-authored context) + 'user', -- A message from the account + 'assistant' -- A reply from the model +); + +COMMENT ON TYPE CHAT_ROLE IS 'Author of a chat message: system, user, or assistant.'; + +-- Chat sessions table: one conversation thread within a workspace. +CREATE TABLE chat_sessions ( + -- Primary identifier + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + + -- References + workspace_id UUID NOT NULL REFERENCES workspaces (id) ON DELETE CASCADE, + account_id UUID NOT NULL REFERENCES accounts (id) ON DELETE CASCADE, + + -- Human-readable title, seeded from the first message (editable). + title TEXT NOT NULL DEFAULT 'New chat', + CONSTRAINT chat_sessions_title_length CHECK (length(trim(title)) BETWEEN 1 AND 255), + + -- The active leaf of the message tree: the message this conversation + -- currently ends at. A client resumes from here, and a new turn without an + -- explicit parent extends this. The FK is added after chat_messages exists. + current_message_id UUID DEFAULT NULL, + + -- Lifecycle timestamps + created_at TIMESTAMPTZ NOT NULL DEFAULT current_timestamp, + updated_at TIMESTAMPTZ NOT NULL DEFAULT current_timestamp, + deleted_at TIMESTAMPTZ DEFAULT NULL, + CONSTRAINT chat_sessions_updated_after_created CHECK (updated_at >= created_at), + CONSTRAINT chat_sessions_deleted_after_created CHECK (deleted_at IS NULL OR deleted_at >= created_at), + CONSTRAINT chat_sessions_deleted_after_updated CHECK (deleted_at IS NULL OR deleted_at >= updated_at) +); + +-- Most recent live sessions per workspace (the session list). +CREATE INDEX chat_sessions_workspace_recent_idx + ON chat_sessions (workspace_id, updated_at DESC) + WHERE deleted_at IS NULL; + +COMMENT ON TABLE chat_sessions IS 'Workspace-scoped assistant conversation threads.'; +COMMENT ON COLUMN chat_sessions.id IS 'Unique session identifier'; +COMMENT ON COLUMN chat_sessions.workspace_id IS 'Workspace this session belongs to'; +COMMENT ON COLUMN chat_sessions.account_id IS 'Account that opened the session'; +COMMENT ON COLUMN chat_sessions.title IS 'Human-readable title (seeded from the first message)'; +COMMENT ON COLUMN chat_sessions.current_message_id IS 'Active leaf of the message tree (resume point)'; +COMMENT ON COLUMN chat_sessions.created_at IS 'Session creation timestamp'; +COMMENT ON COLUMN chat_sessions.updated_at IS 'Timestamp of the most recent message'; +COMMENT ON COLUMN chat_sessions.deleted_at IS 'Soft-deletion timestamp; NULL means live'; + +-- Chat messages table: the conversation tree of a session. +CREATE TABLE chat_messages ( + -- Primary identifier + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + + -- References + session_id UUID NOT NULL REFERENCES chat_sessions (id) ON DELETE CASCADE, + + -- Composite key target: lets the tree/leaf foreign keys pin a message to a + -- specific session, so a parent (or a session's active leaf) can never point + -- at a message from another session. + CONSTRAINT chat_messages_id_session_key UNIQUE (id, session_id), + + -- The message this one replies to (its parent in the conversation tree). + -- NULL is a root. A regenerated reply is a sibling: another child of the same + -- parent. The active conversation is the path from a leaf back to the root. + -- The composite FK enforces that a parent is in the same session. + parent_id UUID DEFAULT NULL, + CONSTRAINT chat_messages_parent_fkey + FOREIGN KEY (parent_id, session_id) + REFERENCES chat_messages (id, session_id) ON DELETE CASCADE, + + -- Message details. The content is stored XChaCha20-Poly1305 encrypted with + -- the workspace-derived key (a user may paste sensitive text into the + -- assistant), so it is opaque bytes rather than searchable text. + role CHAT_ROLE NOT NULL, + content BYTEA NOT NULL, + CONSTRAINT chat_messages_content_size CHECK (length(content) BETWEEN 1 AND 131072), + + -- Lifecycle timestamp + created_at TIMESTAMPTZ NOT NULL DEFAULT current_timestamp +); + +-- All of a session's messages (the whole tree; the path is walked in-app). +CREATE INDEX chat_messages_session_idx + ON chat_messages (session_id); + +-- Walk a node's children (sibling branches), and back the parent composite FK. +CREATE INDEX chat_messages_parent_idx + ON chat_messages (parent_id, session_id) + WHERE parent_id IS NOT NULL; + +-- The active-leaf pointer references a message in THIS session: the composite FK +-- ties the session's own id to the referenced message's session_id. Added now +-- that chat_messages exists. A deleted leaf clears the pointer rather than +-- cascading the session. +ALTER TABLE chat_sessions + ADD CONSTRAINT chat_sessions_current_message_fkey + FOREIGN KEY (current_message_id, id) + REFERENCES chat_messages (id, session_id) + ON DELETE SET NULL (current_message_id); + +COMMENT ON TABLE chat_messages IS 'Messages of a chat session, as a conversation tree.'; +COMMENT ON COLUMN chat_messages.id IS 'Unique message identifier'; +COMMENT ON COLUMN chat_messages.session_id IS 'Session this message belongs to'; +COMMENT ON COLUMN chat_messages.parent_id IS 'Parent in the conversation tree; NULL is a root'; +COMMENT ON COLUMN chat_messages.role IS 'Author of the message (system, user, or assistant)'; +COMMENT ON COLUMN chat_messages.content IS 'XChaCha20-Poly1305 encrypted message text'; +COMMENT ON COLUMN chat_messages.created_at IS 'Message creation timestamp';