diff --git a/crates/nvisy-postgres/src/model/workspace_connection.rs b/crates/nvisy-postgres/src/model/workspace_connection.rs index e313abc..f7ccb2d 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, Slug}; +use crate::types::{HasCreatedAt, HasDeletedAt, HasUpdatedAt}; /// Workspace connection model representing encrypted provider connections. /// @@ -22,8 +22,6 @@ pub struct WorkspaceConnection { pub workspace_id: Uuid, /// Reference to the account that created this connection. pub account_id: Uuid, - /// URL-safe connection identifier, unique within the workspace. - pub slug: Slug, /// Human-readable connection name. pub name: String, /// Provider type for indexing (e.g., "openai", "postgres", "s3"). @@ -52,8 +50,6 @@ pub struct NewWorkspaceConnection { pub workspace_id: Uuid, /// Account ID (required). pub account_id: Uuid, - /// URL-safe connection identifier, unique within the workspace. - pub slug: Slug, /// Connection name. pub name: String, /// Provider type for indexing. diff --git a/crates/nvisy-postgres/src/model/workspace_connection_run.rs b/crates/nvisy-postgres/src/model/workspace_connection_run.rs index 2b5bfb8..b8d1155 100644 --- a/crates/nvisy-postgres/src/model/workspace_connection_run.rs +++ b/crates/nvisy-postgres/src/model/workspace_connection_run.rs @@ -28,8 +28,6 @@ pub struct WorkspaceConnectionRun { pub trigger_type: SyncTriggerType, /// Current run status. pub status: SyncStatus, - /// Human-facing sequence number within the connection (assigned at insert). - pub run_number: i32, /// Number of records processed. pub records_synced: i64, /// Failure detail when status is failed. diff --git a/crates/nvisy-postgres/src/model/workspace_pipeline_run.rs b/crates/nvisy-postgres/src/model/workspace_pipeline_run.rs index 2ffec5d..adc94ef 100644 --- a/crates/nvisy-postgres/src/model/workspace_pipeline_run.rs +++ b/crates/nvisy-postgres/src/model/workspace_pipeline_run.rs @@ -28,8 +28,6 @@ pub struct WorkspacePipelineRun { pub trigger_type: PipelineTriggerType, /// Current run status. pub status: PipelineRunStatus, - /// Human-facing sequence number within the pipeline (assigned at insert). - pub run_number: i32, /// Object-store key for the encrypted `AnalyzedDocument` held between detect /// and redact. `None` until analysis writes it. pub analyzed_document_key: Option, diff --git a/crates/nvisy-postgres/src/model/workspace_webhook.rs b/crates/nvisy-postgres/src/model/workspace_webhook.rs index d4a78cb..1aa23d6 100644 --- a/crates/nvisy-postgres/src/model/workspace_webhook.rs +++ b/crates/nvisy-postgres/src/model/workspace_webhook.rs @@ -11,7 +11,7 @@ use uuid::Uuid; use crate::schema::workspace_webhooks; use crate::types::{ - HasCreatedAt, HasDeletedAt, HasOwnership, HasUpdatedAt, Slug, WebhookEvent, WebhookStatus, + HasCreatedAt, HasDeletedAt, HasOwnership, HasUpdatedAt, WebhookEvent, WebhookStatus, }; /// Workspace webhook model representing a webhook configuration for a workspace. @@ -27,8 +27,6 @@ pub struct WorkspaceWebhook { pub id: Uuid, /// Reference to the workspace this webhook belongs to. pub workspace_id: Uuid, - /// URL-safe webhook identifier, unique within the workspace. - pub slug: Slug, /// Human-readable name for the webhook. pub display_name: String, /// Description of the webhook's purpose. @@ -62,8 +60,6 @@ pub struct WorkspaceWebhook { pub struct NewWorkspaceWebhook { /// Reference to the workspace this webhook will belong to. pub workspace_id: Uuid, - /// URL-safe webhook identifier, unique within the workspace. - pub slug: Slug, /// Human-readable name for the webhook. pub display_name: String, /// Description of the webhook's purpose. diff --git a/crates/nvisy-postgres/src/query/workspace_connection.rs b/crates/nvisy-postgres/src/query/workspace_connection.rs index 6e9f680..c897a47 100644 --- a/crates/nvisy-postgres/src/query/workspace_connection.rs +++ b/crates/nvisy-postgres/src/query/workspace_connection.rs @@ -36,14 +36,14 @@ pub trait WorkspaceConnectionRepository { connection_id: Uuid, ) -> impl Future>> + Send; - /// Finds a connection by slug within a specific workspace, with the handle - /// of the account that created it. + /// Finds a connection by id within a specific workspace, with the handle of + /// the account that created it. /// /// Excludes soft-deleted connections. - fn find_connection_in_workspace_by_slug( + fn find_connection_in_workspace_with_creator( &mut self, workspace_id: Uuid, - slug: &str, + connection_id: Uuid, ) -> impl Future>> + Send; /// Finds connections by provider type within a workspace. @@ -151,10 +151,10 @@ impl WorkspaceConnectionRepository for PgConnection { Ok(connection) } - async fn find_connection_in_workspace_by_slug( + async fn find_connection_in_workspace_with_creator( &mut self, workspace_id: Uuid, - slug: &str, + connection_id: Uuid, ) -> PgResult> { use schema::workspace_connections::dsl; use schema::{accounts, workspace_connections}; @@ -162,7 +162,7 @@ impl WorkspaceConnectionRepository for PgConnection { let connection = workspace_connections::table .inner_join(accounts::table) .filter(dsl::workspace_id.eq(workspace_id)) - .filter(dsl::slug.eq(slug)) + .filter(dsl::id.eq(connection_id)) .filter(dsl::deleted_at.is_null()) .select((WorkspaceConnection::as_select(), accounts::username)) .first(self) diff --git a/crates/nvisy-postgres/src/query/workspace_connection_run.rs b/crates/nvisy-postgres/src/query/workspace_connection_run.rs index ee10b7c..0bd99fc 100644 --- a/crates/nvisy-postgres/src/query/workspace_connection_run.rs +++ b/crates/nvisy-postgres/src/query/workspace_connection_run.rs @@ -9,7 +9,7 @@ use uuid::Uuid; use crate::model::{ NewWorkspaceConnectionRun, UpdateWorkspaceConnectionRun, WorkspaceConnectionRun, }; -use crate::types::{CursorPage, CursorPagination, Slug, SyncStatus, Username}; +use crate::types::{CursorPage, CursorPagination, SyncStatus, Username}; use crate::{PgConnection, PgError, PgResult, schema}; /// Repository for workspace connection run database operations. @@ -40,16 +40,30 @@ pub trait WorkspaceConnectionRunRepository { run_id: Uuid, ) -> impl Future>> + Send; - /// Finds a run by its per-connection sequential `run_number`. + /// Finds a sync run by its opaque id, scoped to a workspace via its owning + /// connection, with the triggering account's handle. /// - /// The run's public identity is `(connection, run_number)`; this is the - /// lookup behind `/connections/{slug}/runs/{run_number}`. - fn find_connection_run_by_number( + /// Runs carry no workspace column, so this joins through the connection and + /// filters on its workspace; a run whose connection is in another workspace + /// is not found. + fn find_connection_run_by_id( &mut self, - connection_id: Uuid, - run_number: i32, + workspace_id: Uuid, + run_id: Uuid, ) -> impl Future)>>> + Send; + /// Returns the most recent successful sync completion time for each of the + /// given connections. + /// + /// A connection's "last synced" instant is the `completed_at` of its latest + /// run with status `Completed`; connections that have never synced + /// successfully are absent from the result. This is a single grouped query + /// so a page of connections costs one round-trip, not one per connection. + fn last_successful_sync_at( + &mut self, + connection_ids: &[Uuid], + ) -> impl Future>> + Send; + /// Lists runs for a specific connection with cursor pagination, each paired /// with the handle of the account that triggered it, if any. fn cursor_list_workspace_connection_runs( @@ -72,7 +86,7 @@ pub trait WorkspaceConnectionRunRepository { workspace_id: Uuid, pagination: CursorPagination, status_filter: Option, - ) -> impl Future)>>> + ) -> impl Future)>>> + Send; /// Gets the most recent run for a connection (its current sync state). @@ -163,18 +177,21 @@ impl WorkspaceConnectionRunRepository for PgConnection { Ok(run) } - async fn find_connection_run_by_number( + async fn find_connection_run_by_id( &mut self, - connection_id: Uuid, - run_number: i32, + workspace_id: Uuid, + run_id: Uuid, ) -> PgResult)>> { - use schema::workspace_connection_runs::dsl; - use schema::{accounts, workspace_connection_runs}; + use schema::workspace_connection_runs::dsl as runs; + use schema::{accounts, workspace_connection_runs, workspace_connections}; + // Runs carry no workspace column; scope through the owning connection so + // the id resolves only within its workspace. let run = workspace_connection_runs::table + .inner_join(workspace_connections::table) .left_join(accounts::table) - .filter(dsl::connection_id.eq(connection_id)) - .filter(dsl::run_number.eq(run_number)) + .filter(runs::id.eq(run_id)) + .filter(workspace_connections::workspace_id.eq(workspace_id)) .select(( WorkspaceConnectionRun::as_select(), accounts::username.nullable(), @@ -187,6 +204,30 @@ impl WorkspaceConnectionRunRepository for PgConnection { Ok(run) } + async fn last_successful_sync_at( + &mut self, + connection_ids: &[Uuid], + ) -> PgResult> { + use diesel::dsl::max; + use schema::workspace_connection_runs::{self, dsl}; + + if connection_ids.is_empty() { + return Ok(Vec::new()); + } + + // Only successful runs count toward "last synced"; a failed or cancelled + // run does not move the timestamp. completed_at is non-null for any run + // in a terminal state, so the grouped MAX is present for every group. + workspace_connection_runs::table + .filter(dsl::connection_id.eq_any(connection_ids)) + .filter(dsl::status.eq(SyncStatus::Completed)) + .group_by(dsl::connection_id) + .select((dsl::connection_id, max(dsl::completed_at).assume_not_null())) + .load(self) + .await + .map_err(PgError::from) + } + async fn cursor_list_workspace_connection_runs( &mut self, connection_id: Uuid, @@ -272,15 +313,15 @@ impl WorkspaceConnectionRunRepository for PgConnection { workspace_id: Uuid, pagination: CursorPagination, status_filter: Option, - ) -> PgResult)>> { + ) -> PgResult)>> { use schema::accounts::dsl as accounts; use schema::workspace_connection_runs::dsl as runs; use schema::workspace_connections::dsl as connections; // Runs have no workspace column; scope them through the owning - // connection. The owning connection's slug and the triggering account's + // connection. The owning connection's id and the triggering account's // handle are selected alongside each run so the cross-connection response - // can address each run by `(connection, number)` and name its trigger. + // can name its connection and trigger (the run is addressed by its own id). let scoped = || { let mut query = runs::workspace_connection_runs .inner_join(connections::workspace_connections) @@ -308,11 +349,11 @@ impl WorkspaceConnectionRunRepository for PgConnection { let limit = pagination.limit + 1; let selection = ( WorkspaceConnectionRun::as_select(), - connections::slug, + connections::id, accounts::username.nullable(), ); - let items: Vec<(WorkspaceConnectionRun, Slug, Option)> = + let items: Vec<(WorkspaceConnectionRun, Uuid, Option)> = if let Some(cursor) = &pagination.after { let cursor_time = jiff_diesel::Timestamp::from(cursor.timestamp); @@ -342,7 +383,7 @@ impl WorkspaceConnectionRunRepository for PgConnection { items, total, pagination.limit, - |(run, _, _): &(WorkspaceConnectionRun, Slug, Option)| { + |(run, _, _): &(WorkspaceConnectionRun, Uuid, Option)| { (run.started_at.into(), run.id) }, )) diff --git a/crates/nvisy-postgres/src/query/workspace_pipeline_run.rs b/crates/nvisy-postgres/src/query/workspace_pipeline_run.rs index f79ccec..cd0b07f 100644 --- a/crates/nvisy-postgres/src/query/workspace_pipeline_run.rs +++ b/crates/nvisy-postgres/src/query/workspace_pipeline_run.rs @@ -6,7 +6,9 @@ use diesel::prelude::*; use diesel_async::RunQueryDsl; use uuid::Uuid; -use crate::model::{NewWorkspacePipelineRun, UpdateWorkspacePipelineRun, WorkspacePipelineRun}; +use crate::model::{ + NewWorkspacePipelineRun, UpdateWorkspacePipelineRun, WorkspacePipeline, WorkspacePipelineRun, +}; use crate::types::{ CursorPage, CursorPagination, OffsetPagination, PipelineRunStatus, Slug, Username, }; @@ -23,32 +25,19 @@ pub trait WorkspacePipelineRunRepository { new_run: NewWorkspacePipelineRun, ) -> impl Future> + Send; - /// Finds a workspace pipeline run by its unique identifier. - fn find_workspace_pipeline_run_by_id( - &mut self, - run_id: Uuid, - ) -> impl Future>> + Send; - - /// Finds a run by ID, scoped to a workspace via its owning pipeline. + /// Finds a run by its opaque id, scoped to a workspace, returning the run, + /// its owning pipeline, and the triggering account's handle. /// - /// Runs carry no workspace column, so this joins through the pipeline and - /// filters on its workspace. A run whose pipeline is in another workspace - /// is not found. - fn find_pipeline_run_in_workspace( + /// The run is addressed by its own id (behind `/runs/{runId}`); scoping + /// through the owning pipeline keeps it workspace-bounded and hides runs of + /// soft-deleted pipelines. + fn find_workspace_run_by_id( &mut self, workspace_id: Uuid, run_id: Uuid, - ) -> impl Future>> + Send; - - /// Finds a run by its per-pipeline sequential `run_number`. - /// - /// The run's public identity is `(pipeline, run_number)`; this is the - /// lookup behind `/pipelines/{slug}/runs/{run_number}`. - fn find_pipeline_run_by_number( - &mut self, - pipeline_id: Uuid, - run_number: i32, - ) -> impl Future)>>> + Send; + ) -> impl Future< + Output = PgResult)>>, + > + Send; /// Finds a run by its `(pipeline, idempotency key)` pair, for detect replay. fn find_pipeline_run_by_idempotency_key( @@ -156,58 +145,27 @@ impl WorkspacePipelineRunRepository for PgConnection { Ok(run) } - async fn find_workspace_pipeline_run_by_id( - &mut self, - run_id: Uuid, - ) -> PgResult> { - use schema::workspace_pipeline_runs::{self, dsl}; - - let run = workspace_pipeline_runs::table - .filter(dsl::id.eq(run_id)) - .select(WorkspacePipelineRun::as_select()) - .first(self) - .await - .optional() - .map_err(PgError::from)?; - - Ok(run) - } - - async fn find_pipeline_run_in_workspace( + async fn find_workspace_run_by_id( &mut self, workspace_id: Uuid, run_id: Uuid, - ) -> PgResult> { + ) -> PgResult)>> { use schema::workspace_pipeline_runs::dsl as runs; - use schema::workspace_pipelines::dsl as pipelines; - - let run = runs::workspace_pipeline_runs - .inner_join(pipelines::workspace_pipelines) - .filter(runs::id.eq(run_id)) - .filter(pipelines::workspace_id.eq(workspace_id)) - .select(WorkspacePipelineRun::as_select()) - .first(self) - .await - .optional() - .map_err(PgError::from)?; - - Ok(run) - } - - async fn find_pipeline_run_by_number( - &mut self, - pipeline_id: Uuid, - run_number: i32, - ) -> PgResult)>> { - use schema::workspace_pipeline_runs::dsl; - use schema::{accounts, workspace_pipeline_runs}; + use schema::{accounts, workspace_pipeline_runs, workspace_pipelines}; + // Runs carry no workspace column; scope through the owning pipeline so + // the id resolves only within its workspace, and only while that + // pipeline is live (a soft-deleted pipeline hides its runs). The + // pipeline is returned alongside so callers need no second lookup. let run = workspace_pipeline_runs::table + .inner_join(workspace_pipelines::table) .left_join(accounts::table) - .filter(dsl::pipeline_id.eq(pipeline_id)) - .filter(dsl::run_number.eq(run_number)) + .filter(runs::id.eq(run_id)) + .filter(workspace_pipelines::workspace_id.eq(workspace_id)) + .filter(workspace_pipelines::deleted_at.is_null()) .select(( WorkspacePipelineRun::as_select(), + WorkspacePipeline::as_select(), accounts::username.nullable(), )) .first(self) @@ -351,8 +309,8 @@ impl WorkspacePipelineRunRepository for PgConnection { // Runs have no workspace column; scope them through the owning pipeline. // The owning pipeline's slug and the triggering account's handle are - // selected alongside each run so the cross-pipeline response can address - // each run by `(pipeline, number)` and name its trigger. + // selected alongside each run so the cross-pipeline response can name + // its pipeline and trigger (the run is addressed by its own id). let scoped = || { let mut query = runs::workspace_pipeline_runs .inner_join(pipelines::workspace_pipelines) diff --git a/crates/nvisy-postgres/src/query/workspace_webhook.rs b/crates/nvisy-postgres/src/query/workspace_webhook.rs index ec34a4f..19c531d 100644 --- a/crates/nvisy-postgres/src/query/workspace_webhook.rs +++ b/crates/nvisy-postgres/src/query/workspace_webhook.rs @@ -35,12 +35,12 @@ pub trait WorkspaceWebhookRepository { webhook_id: Uuid, ) -> impl Future>> + Send; - /// Finds a webhook by slug within a workspace, with the handle of the - /// account that created it, excluding soft-deleted rows. - fn find_webhook_in_workspace_by_slug( + /// Finds a webhook by id within a workspace, with the handle of the account + /// that created it, excluding soft-deleted rows. + fn find_webhook_in_workspace_with_creator( &mut self, workspace_id: Uuid, - slug: &str, + webhook_id: Uuid, ) -> impl Future>> + Send; /// Lists all webhooks for a workspace with offset pagination. @@ -170,17 +170,17 @@ impl WorkspaceWebhookRepository for PgConnection { Ok(webhook) } - async fn find_webhook_in_workspace_by_slug( + async fn find_webhook_in_workspace_with_creator( &mut self, workspace_id: Uuid, - slug_value: &str, + webhook_id: Uuid, ) -> PgResult> { use schema::workspace_webhooks::dsl; use schema::{accounts, workspace_webhooks}; let webhook = workspace_webhooks::table .inner_join(accounts::table) - .filter(dsl::slug.eq(slug_value)) + .filter(dsl::id.eq(webhook_id)) .filter(dsl::workspace_id.eq(workspace_id)) .filter(dsl::deleted_at.is_null()) .select((WorkspaceWebhook::as_select(), accounts::username)) diff --git a/crates/nvisy-postgres/src/schema.rs b/crates/nvisy-postgres/src/schema.rs index aaf5161..16b5e93 100644 --- a/crates/nvisy-postgres/src/schema.rs +++ b/crates/nvisy-postgres/src/schema.rs @@ -147,7 +147,6 @@ diesel::table! { account_id -> Nullable, trigger_type -> SyncTriggerType, status -> SyncStatus, - run_number -> Int4, records_synced -> Int8, error_message -> Nullable, metadata -> Jsonb, @@ -163,7 +162,6 @@ diesel::table! { id -> Uuid, workspace_id -> Uuid, account_id -> Uuid, - slug -> Text, name -> Text, provider -> Text, encrypted_data -> Bytea, @@ -307,7 +305,6 @@ diesel::table! { account_id -> Nullable, trigger_type -> PipelineTriggerType, status -> PipelineRunStatus, - run_number -> Int4, analyzed_document_key -> Nullable, idempotency_key -> Nullable, metadata -> Jsonb, @@ -366,7 +363,6 @@ diesel::table! { workspace_webhooks (id) { id -> Uuid, workspace_id -> Uuid, - slug -> Text, display_name -> Text, description -> Text, url -> Text, diff --git a/crates/nvisy-postgres/src/types/constraint/mod.rs b/crates/nvisy-postgres/src/types/constraint/mod.rs index 597d4ca..9ef2c2b 100644 --- a/crates/nvisy-postgres/src/types/constraint/mod.rs +++ b/crates/nvisy-postgres/src/types/constraint/mod.rs @@ -335,13 +335,6 @@ mod tests { WorkspacePolicyConstraints::WorkspaceIdIdUnique )) ); - assert_eq!( - ConstraintViolation::new("workspace_pipeline_runs_pipeline_id_run_number_key"), - Some(ConstraintViolation::WorkspacePipelineRun( - WorkspacePipelineRunConstraints::RunNumberUnique - )) - ); - assert_eq!(ConstraintViolation::new("unknown_constraint"), None); } @@ -390,8 +383,8 @@ mod tests { ConstraintCategory::Chronological ); - let violation = ConstraintViolation::WorkspacePipelineRun( - WorkspacePipelineRunConstraints::RunNumberUnique, + let violation = ConstraintViolation::WorkspaceConnection( + WorkspaceConnectionConstraints::WorkspaceIdIdUnique, ); assert_eq!( violation.constraint_category(), diff --git a/crates/nvisy-postgres/src/types/constraint/pipeline_runs.rs b/crates/nvisy-postgres/src/types/constraint/pipeline_runs.rs index 2fbf8f0..b7b22e8 100644 --- a/crates/nvisy-postgres/src/types/constraint/pipeline_runs.rs +++ b/crates/nvisy-postgres/src/types/constraint/pipeline_runs.rs @@ -17,12 +17,6 @@ pub enum WorkspacePipelineRunConstraints { MetadataSize, #[strum(serialize = "workspace_pipeline_runs_idempotency_key_length")] IdempotencyKeyLength, - #[strum(serialize = "workspace_pipeline_runs_run_number_positive")] - RunNumberPositive, - - // Uniqueness constraints - #[strum(serialize = "workspace_pipeline_runs_pipeline_id_run_number_key")] - RunNumberUnique, // Chronological constraints #[strum(serialize = "workspace_pipeline_runs_completed_after_started")] @@ -40,10 +34,9 @@ impl WorkspacePipelineRunConstraints { match self { WorkspacePipelineRunConstraints::AnalyzedDocumentKeyLength | WorkspacePipelineRunConstraints::MetadataSize - | WorkspacePipelineRunConstraints::IdempotencyKeyLength - | WorkspacePipelineRunConstraints::RunNumberPositive => ConstraintCategory::Validation, - - WorkspacePipelineRunConstraints::RunNumberUnique => ConstraintCategory::Uniqueness, + | WorkspacePipelineRunConstraints::IdempotencyKeyLength => { + ConstraintCategory::Validation + } WorkspacePipelineRunConstraints::CompletedAfterStarted => { ConstraintCategory::Chronological diff --git a/crates/nvisy-postgres/src/types/constraint/workspace_connection_runs.rs b/crates/nvisy-postgres/src/types/constraint/workspace_connection_runs.rs index 9b66b5f..740f135 100644 --- a/crates/nvisy-postgres/src/types/constraint/workspace_connection_runs.rs +++ b/crates/nvisy-postgres/src/types/constraint/workspace_connection_runs.rs @@ -17,12 +17,6 @@ pub enum WorkspaceConnectionRunConstraints { ErrorMessageLength, #[strum(serialize = "workspace_connection_runs_metadata_size")] MetadataSize, - #[strum(serialize = "workspace_connection_runs_run_number_positive")] - RunNumberPositive, - - // Uniqueness constraints - #[strum(serialize = "workspace_connection_runs_connection_id_run_number_key")] - RunNumberUnique, // Chronological constraints #[strum(serialize = "workspace_connection_runs_completed_after_started")] @@ -40,12 +34,7 @@ impl WorkspaceConnectionRunConstraints { match self { WorkspaceConnectionRunConstraints::RecordsSyncedNonNegative | WorkspaceConnectionRunConstraints::ErrorMessageLength - | WorkspaceConnectionRunConstraints::MetadataSize - | WorkspaceConnectionRunConstraints::RunNumberPositive => { - ConstraintCategory::Validation - } - - WorkspaceConnectionRunConstraints::RunNumberUnique => ConstraintCategory::Uniqueness, + | WorkspaceConnectionRunConstraints::MetadataSize => ConstraintCategory::Validation, WorkspaceConnectionRunConstraints::CompletedAfterStarted => { ConstraintCategory::Chronological diff --git a/crates/nvisy-postgres/src/types/constraint/workspace_connections.rs b/crates/nvisy-postgres/src/types/constraint/workspace_connections.rs index 33c13d8..77d63f5 100644 --- a/crates/nvisy-postgres/src/types/constraint/workspace_connections.rs +++ b/crates/nvisy-postgres/src/types/constraint/workspace_connections.rs @@ -10,12 +10,6 @@ use super::ConstraintCategory; #[derive(Serialize, Deserialize, Display, EnumIter, EnumString)] #[serde(into = "String", try_from = "String")] pub enum WorkspaceConnectionConstraints { - // Slug validation constraints - #[strum(serialize = "workspace_connections_slug_length")] - SlugLength, - #[strum(serialize = "workspace_connections_slug_format")] - SlugFormat, - // Name validation constraints #[strum(serialize = "workspace_connections_name_length")] NameLength, @@ -35,8 +29,8 @@ pub enum WorkspaceConnectionConstraints { // Uniqueness constraints #[strum(serialize = "workspace_connections_workspace_id_id_key")] WorkspaceIdIdUnique, - #[strum(serialize = "workspace_connections_workspace_id_slug_key")] - SlugUnique, + #[strum(serialize = "workspace_connections_name_unique_idx")] + NameUnique, // Chronological constraints #[strum(serialize = "workspace_connections_updated_after_created")] @@ -54,15 +48,13 @@ impl WorkspaceConnectionConstraints { /// Returns the category of this constraint violation. pub fn categorize(&self) -> ConstraintCategory { match self { - WorkspaceConnectionConstraints::SlugLength - | WorkspaceConnectionConstraints::SlugFormat - | WorkspaceConnectionConstraints::NameLength + WorkspaceConnectionConstraints::NameLength | WorkspaceConnectionConstraints::ProviderLength | WorkspaceConnectionConstraints::DataSize | WorkspaceConnectionConstraints::MetadataSize => ConstraintCategory::Validation, WorkspaceConnectionConstraints::WorkspaceIdIdUnique - | WorkspaceConnectionConstraints::SlugUnique => ConstraintCategory::Uniqueness, + | WorkspaceConnectionConstraints::NameUnique => ConstraintCategory::Uniqueness, WorkspaceConnectionConstraints::UpdatedAfterCreated | WorkspaceConnectionConstraints::DeletedAfterCreated => { diff --git a/crates/nvisy-postgres/src/types/constraint/workspace_webhooks.rs b/crates/nvisy-postgres/src/types/constraint/workspace_webhooks.rs index 37a4e85..b749b57 100644 --- a/crates/nvisy-postgres/src/types/constraint/workspace_webhooks.rs +++ b/crates/nvisy-postgres/src/types/constraint/workspace_webhooks.rs @@ -13,14 +13,8 @@ pub enum WorkspaceWebhookConstraints { // Webhook unique constraints #[strum(serialize = "workspace_webhooks_workspace_id_id_key")] WorkspaceIdIdUnique, - #[strum(serialize = "workspace_webhooks_workspace_id_slug_key")] - SlugUnique, // Webhook validation constraints - #[strum(serialize = "workspace_webhooks_slug_length")] - SlugLength, - #[strum(serialize = "workspace_webhooks_slug_format")] - SlugFormat, #[strum(serialize = "workspace_webhooks_display_name_length")] DisplayNameLength, #[strum(serialize = "workspace_webhooks_description_length")] @@ -50,12 +44,9 @@ impl WorkspaceWebhookConstraints { /// Returns the category of this constraint violation. pub fn categorize(&self) -> ConstraintCategory { match self { - WorkspaceWebhookConstraints::WorkspaceIdIdUnique - | WorkspaceWebhookConstraints::SlugUnique => ConstraintCategory::Uniqueness, + WorkspaceWebhookConstraints::WorkspaceIdIdUnique => ConstraintCategory::Uniqueness, - WorkspaceWebhookConstraints::SlugLength - | WorkspaceWebhookConstraints::SlugFormat - | WorkspaceWebhookConstraints::DisplayNameLength + WorkspaceWebhookConstraints::DisplayNameLength | WorkspaceWebhookConstraints::DescriptionLength | WorkspaceWebhookConstraints::UrlLength | WorkspaceWebhookConstraints::UrlFormat diff --git a/crates/nvisy-postgres/src/types/mod.rs b/crates/nvisy-postgres/src/types/mod.rs index 8ea53f2..42410b2 100644 --- a/crates/nvisy-postgres/src/types/mod.rs +++ b/crates/nvisy-postgres/src/types/mod.rs @@ -5,6 +5,7 @@ mod constraint; mod enums; mod filtering; mod pagination; +mod prefixed_id; mod slug; mod sorting; mod username; @@ -30,6 +31,7 @@ pub use enums::{ }; pub use filtering::{FileFilter, FileFormat, InviteFilter, MemberFilter}; pub use pagination::{Cursor, CursorPage, CursorPagination, OffsetPage, OffsetPagination}; +pub use prefixed_id::{ConnectionId, PrefixedIdError, RunId, WebhookId}; pub use slug::{SLUG_MAX_LENGTH, SLUG_MIN_LENGTH, Slug, SlugError}; pub use sorting::{ FileSortBy, FileSortField, InviteSortBy, InviteSortField, MemberSortBy, MemberSortField, diff --git a/crates/nvisy-postgres/src/types/prefixed_id.rs b/crates/nvisy-postgres/src/types/prefixed_id.rs new file mode 100644 index 0000000..effb46c --- /dev/null +++ b/crates/nvisy-postgres/src/types/prefixed_id.rs @@ -0,0 +1,185 @@ +//! Prefixed identifiers: type-prefixed opaque ids for API resources. + +use std::fmt; +use std::str::FromStr; + +use uuid::Uuid; + +/// Error returned when a string is not a valid prefixed id. +#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] +pub enum PrefixedIdError { + /// The value does not start with the expected `_` marker. + #[error("id must start with the '{0}_' prefix")] + Prefix(&'static str), + /// The portion after the prefix is not a valid UUID. + #[error("id does not contain a valid identifier")] + Uuid, +} + +/// Declares a distinct, type-prefixed opaque id newtype wrapping a [`Uuid`]. +/// +/// The id renders as `_` at the API boundary and parses the same +/// shape back. The underlying database column remains `Uuid`; the prefix is a +/// presentation encoding only, so these types convert to and from [`Uuid`] at +/// the handler edge and are never used as a Diesel SQL type. Each invocation +/// produces its own type, so ids of different resources cannot be interchanged. +macro_rules! prefixed_id { + ($(#[$meta:meta])* $name:ident, $prefix:literal) => { + $(#[$meta])* + #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] + #[derive(serde::Serialize, serde::Deserialize)] + #[serde(try_from = "String", into = "String")] + pub struct $name(Uuid); + + impl $name { + /// The textual prefix (without the trailing underscore). + pub const PREFIX: &'static str = $prefix; + + /// Wraps a raw [`Uuid`] as this prefixed id. + #[inline] + #[must_use] + pub const fn from_uuid(id: Uuid) -> Self { + Self(id) + } + + /// Returns the underlying [`Uuid`]. + #[inline] + #[must_use] + pub const fn as_uuid(&self) -> Uuid { + self.0 + } + + /// Parses a `_` string into this id. + /// + /// # Errors + /// + /// Returns [`PrefixedIdError`] if the prefix is wrong or the + /// remainder is not a valid UUID. + pub fn parse(value: &str) -> Result { + let rest = value + .strip_prefix(concat!($prefix, "_")) + .ok_or(PrefixedIdError::Prefix($prefix))?; + let id = Uuid::parse_str(rest).map_err(|_| PrefixedIdError::Uuid)?; + Ok(Self(id)) + } + } + + impl fmt::Display for $name { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}_{}", $prefix, self.0) + } + } + + impl From for $name { + #[inline] + fn from(id: Uuid) -> Self { + Self(id) + } + } + + impl From<$name> for Uuid { + #[inline] + fn from(id: $name) -> Self { + id.0 + } + } + + impl FromStr for $name { + type Err = PrefixedIdError; + + #[inline] + fn from_str(value: &str) -> Result { + Self::parse(value) + } + } + + impl TryFrom for $name { + type Error = PrefixedIdError; + + #[inline] + fn try_from(value: String) -> Result { + Self::parse(&value) + } + } + + impl From<$name> for String { + #[inline] + fn from(id: $name) -> Self { + id.to_string() + } + } + + #[cfg(feature = "schema")] + impl schemars::JsonSchema for $name { + fn schema_name() -> std::borrow::Cow<'static, str> { + stringify!($name).into() + } + + fn json_schema(_generator: &mut schemars::SchemaGenerator) -> schemars::Schema { + schemars::json_schema!({ + "type": "string", + "pattern": concat!("^", $prefix, r"_[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$"), + "description": concat!("Opaque ", $prefix, " identifier (", $prefix, "_)."), + }) + } + } + }; +} + +prefixed_id! { + /// Opaque identifier for a workspace connection (`conn_`). + ConnectionId, "conn" +} + +prefixed_id! { + /// Opaque identifier for a workspace webhook (`whk_`). + WebhookId, "whk" +} + +prefixed_id! { + /// Opaque identifier for a pipeline run (`run_`). + RunId, "run" +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn round_trips_through_string() { + let uuid = Uuid::from_u128(0x0123_4567_89ab_cdef_0123_4567_89ab_cdef); + let id = ConnectionId::from_uuid(uuid); + let rendered = id.to_string(); + assert!(rendered.starts_with("conn_")); + assert_eq!(ConnectionId::parse(&rendered).unwrap(), id); + assert_eq!(id.as_uuid(), uuid); + } + + #[test] + fn rejects_wrong_prefix() { + let uuid = Uuid::from_u128(1); + let webhook = WebhookId::from_uuid(uuid).to_string(); + // A whk_ id must not parse as a ConnectionId. + assert_eq!( + ConnectionId::parse(&webhook), + Err(PrefixedIdError::Prefix("conn")) + ); + } + + #[test] + fn rejects_malformed_uuid() { + assert_eq!( + ConnectionId::parse("conn_not-a-uuid"), + Err(PrefixedIdError::Uuid) + ); + } + + #[test] + fn rejects_missing_prefix() { + let bare = Uuid::from_u128(1).to_string(); + assert_eq!( + ConnectionId::parse(&bare), + Err(PrefixedIdError::Prefix("conn")) + ); + } +} diff --git a/crates/nvisy-server/src/handler/connection_runs.rs b/crates/nvisy-server/src/handler/connection_runs.rs deleted file mode 100644 index 867f96f..0000000 --- a/crates/nvisy-server/src/handler/connection_runs.rs +++ /dev/null @@ -1,341 +0,0 @@ -//! Workspace connection sync run handlers. -//! -//! Exposes the sync run history for a connection: the record of each -//! synchronization execution, its trigger, progress, and outcome. A run can be -//! triggered manually; background execution that advances a run to completion -//! is handled elsewhere. - -use aide::axum::ApiRouter; -use aide::transform::TransformOperation; -use axum::extract::State; -use axum::http::StatusCode; -use nvisy_postgres::model::{ - NewWorkspaceConnectionRun, WorkspaceConnection, WorkspaceConnectionRun, -}; -use nvisy_postgres::query::{WorkspaceConnectionRepository, WorkspaceConnectionRunRepository}; -use nvisy_postgres::types::Username; -use nvisy_postgres::{PgClient, PgConn}; -use uuid::Uuid; - -use crate::extract::{AuthProvider, AuthState, Json, Path, Permission, Query, WorkspaceContext}; -use crate::handler::request::{ - ConnectionPathParams, ConnectionRunPathParams, ConnectionRunsQuery, CursorPagination, -}; -use crate::handler::response::{ConnectionRun, ConnectionRunsPage, ErrorResponse}; -use crate::handler::{Error, Result}; -use crate::service::ServiceState; - -/// Tracing target for workspace connection run operations. -const TRACING_TARGET: &str = "nvisy_server::handler::connection_runs"; - -/// Triggers a new sync run for a connection. -/// -/// Records a manually-triggered run in the `running` state. Requires -/// `ManageConnections` permission. -#[tracing::instrument( - skip_all, - fields( - account_id = %auth_state.account_id, - workspace_id = %workspace.id, - connection_slug = %path_params.connection_slug, - ) -)] -async fn create_connection_run( - State(pg_client): State, - AuthState(auth_state): AuthState, - WorkspaceContext(workspace): WorkspaceContext, - Path(path_params): Path, -) -> Result<(StatusCode, Json)> { - tracing::debug!(target: TRACING_TARGET, "Triggering connection sync run"); - - let mut conn = pg_client.get_connection().await?; - - auth_state - .authorize_workspace(&mut conn, workspace.id, Permission::ManageConnections) - .await?; - - let connection = find_connection(&mut conn, workspace.id, &path_params.connection_slug).await?; - - let run = conn - .create_workspace_connection_run(NewWorkspaceConnectionRun { - connection_id: connection.id, - account_id: Some(auth_state.account_id), - ..Default::default() - }) - .await?; - - tracing::debug!(target: TRACING_TARGET, run_number = run.run_number, "Connection sync run triggered"); - - let (_, run, trigger_username) = find_connection_run( - &mut conn, - workspace.id, - connection.slug.as_str(), - run.run_number, - ) - .await?; - - Ok(( - StatusCode::CREATED, - Json(ConnectionRun::from_model( - run, - connection.slug, - workspace.slug, - trigger_username, - )), - )) -} - -fn create_connection_run_docs(op: TransformOperation) -> TransformOperation { - op.summary("Trigger connection sync run") - .description("Records a manually-triggered sync run for the connection.") - .response::<201, Json>() - .response::<401, Json>() - .response::<403, Json>() - .response::<404, Json>() -} - -/// Lists sync runs for a connection. -/// -/// Returns the run history in reverse chronological order. Requires -/// `ViewConnections` permission. -#[tracing::instrument( - skip_all, - fields( - account_id = %auth_state.account_id, - workspace_id = %workspace.id, - connection_slug = %path_params.connection_slug, - ) -)] -async fn list_connection_runs( - State(pg_client): State, - AuthState(auth_state): AuthState, - WorkspaceContext(workspace): WorkspaceContext, - Path(path_params): Path, - Query(pagination): Query, - Query(query): Query, -) -> Result<(StatusCode, Json)> { - tracing::debug!(target: TRACING_TARGET, "Listing connection sync runs"); - - let mut conn = pg_client.get_connection().await?; - - auth_state - .authorize_workspace(&mut conn, workspace.id, Permission::ViewConnections) - .await?; - - let connection = find_connection(&mut conn, workspace.id, &path_params.connection_slug).await?; - - let page = conn - .cursor_list_workspace_connection_runs(connection.id, pagination.into(), query.status) - .await?; - - tracing::debug!( - target: TRACING_TARGET, - run_count = page.items.len(), - "Connection sync runs listed", - ); - - Ok(( - StatusCode::OK, - Json(ConnectionRunsPage::from_cursor_page( - page, - |(run, trigger_username)| { - ConnectionRun::from_model( - run, - connection.slug.clone(), - workspace.slug.clone(), - trigger_username, - ) - }, - )), - )) -} - -fn list_connection_runs_docs(op: TransformOperation) -> TransformOperation { - op.summary("List connection sync runs") - .description("Returns the connection's sync run history, most recent first.") - .response::<200, Json>() - .response::<401, Json>() - .response::<403, Json>() - .response::<404, Json>() -} - -/// Lists all sync runs across the workspace's connections. -/// -/// Aggregates runs from every connection in the workspace, most recent first, -/// with an optional status filter. Requires `ViewConnections`. -#[tracing::instrument( - skip_all, - fields( - account_id = %auth_state.account_id, - workspace_id = %workspace.id, - ) -)] -async fn list_workspace_connection_runs( - State(pg_client): State, - AuthState(auth_state): AuthState, - WorkspaceContext(workspace): WorkspaceContext, - Query(pagination): Query, - Query(query): Query, -) -> Result<(StatusCode, Json)> { - tracing::debug!(target: TRACING_TARGET, "Listing workspace connection sync runs"); - - let mut conn = pg_client.get_connection().await?; - - auth_state - .authorize_workspace(&mut conn, workspace.id, Permission::ViewConnections) - .await?; - - let page = conn - .cursor_list_workspace_connection_runs_all(workspace.id, pagination.into(), query.status) - .await?; - - tracing::debug!( - target: TRACING_TARGET, - run_count = page.items.len(), - "Workspace connection sync runs listed", - ); - - Ok(( - StatusCode::OK, - Json(ConnectionRunsPage::from_cursor_page( - page, - |(run, connection_slug, trigger_username)| { - ConnectionRun::from_model( - run, - connection_slug, - workspace.slug.clone(), - trigger_username, - ) - }, - )), - )) -} - -fn list_workspace_connection_runs_docs(op: TransformOperation) -> TransformOperation { - op.summary("List workspace connection sync runs") - .description( - "Returns all sync runs across the workspace's connections, most \ - recent first, with an optional status filter.", - ) - .response::<200, Json>() - .response::<401, Json>() - .response::<403, Json>() - .response::<404, Json>() -} - -/// Retrieves a single connection sync run. -/// -/// The run is addressed as `(connection slug, run number)` within the -/// workspace. Requires `ViewConnections` permission. -#[tracing::instrument( - skip_all, - fields( - account_id = %auth_state.account_id, - workspace_id = %workspace.id, - connection_slug = %path_params.connection_slug, - run_number = path_params.run_number, - ) -)] -async fn read_connection_run( - State(pg_client): State, - AuthState(auth_state): AuthState, - WorkspaceContext(workspace): WorkspaceContext, - Path(path_params): Path, -) -> Result<(StatusCode, Json)> { - tracing::debug!(target: TRACING_TARGET, "Getting connection sync run"); - - let mut conn = pg_client.get_connection().await?; - - auth_state - .authorize_workspace(&mut conn, workspace.id, Permission::ViewConnections) - .await?; - - let (connection, run, trigger_username) = find_connection_run( - &mut conn, - workspace.id, - &path_params.connection_slug, - path_params.run_number, - ) - .await?; - - tracing::debug!(target: TRACING_TARGET, "Connection sync run retrieved"); - - Ok(( - StatusCode::OK, - Json(ConnectionRun::from_model( - run, - connection.slug, - workspace.slug, - trigger_username, - )), - )) -} - -fn read_connection_run_docs(op: TransformOperation) -> TransformOperation { - op.summary("Get connection sync run") - .description("Returns a single connection sync run by its number.") - .response::<200, Json>() - .response::<401, Json>() - .response::<403, Json>() - .response::<404, Json>() -} - -/// Loads a connection within a workspace by slug, mapping a missing row to -/// not-found. -async fn find_connection( - conn: &mut PgConn, - workspace_id: Uuid, - connection_slug: &str, -) -> Result { - conn.find_connection_in_workspace_by_slug(workspace_id, connection_slug) - .await? - .map(|(connection, _)| connection) - .ok_or_else(|| Error::not_found("connection")) -} - -/// Resolves a run addressed as `(connection slug, run number)` within a -/// workspace. -/// -/// Returns both the owning connection and the run, or NotFound if either the -/// connection slug or the run number does not resolve. -async fn find_connection_run( - conn: &mut PgConn, - workspace_id: Uuid, - connection_slug: &str, - run_number: i32, -) -> Result<( - WorkspaceConnection, - WorkspaceConnectionRun, - Option, -)> { - let connection = find_connection(conn, workspace_id, connection_slug).await?; - let (run, trigger_username) = conn - .find_connection_run_by_number(connection.id, run_number) - .await? - .ok_or_else(|| Error::not_found("connection_run"))?; - Ok((connection, run, trigger_username)) -} - -/// Returns routes for workspace connection sync run management. -pub fn routes() -> ApiRouter { - use aide::axum::routing::*; - - ApiRouter::new() - .api_route( - "/workspaces/{workspaceSlug}/connections/runs/", - get_with( - list_workspace_connection_runs, - list_workspace_connection_runs_docs, - ), - ) - .api_route( - "/workspaces/{workspaceSlug}/connections/{connectionSlug}/runs/", - post_with(create_connection_run, create_connection_run_docs) - .get_with(list_connection_runs, list_connection_runs_docs), - ) - .api_route( - "/workspaces/{workspaceSlug}/connections/{connectionSlug}/runs/{runNumber}/", - get_with(read_connection_run, read_connection_run_docs), - ) - .with_path_items(|item| item.tag("Connections")) -} diff --git a/crates/nvisy-server/src/handler/connections.rs b/crates/nvisy-server/src/handler/connections.rs index cf526fb..95522b4 100644 --- a/crates/nvisy-server/src/handler/connections.rs +++ b/crates/nvisy-server/src/handler/connections.rs @@ -11,6 +11,8 @@ //! keys (HKDF-SHA256 with XChaCha20-Poly1305). The encrypted data is stored in //! the database and never exposed through the API. +use std::collections::HashMap; + use aide::axum::ApiRouter; use aide::transform::TransformOperation; use axum::extract::State; @@ -18,8 +20,8 @@ use axum::http::StatusCode; use nvisy_postgres::model::{ NewWorkspaceConnection, UpdateWorkspaceConnection, WorkspaceConnection, }; -use nvisy_postgres::query::WorkspaceConnectionRepository; -use nvisy_postgres::types::Username; +use nvisy_postgres::query::{WorkspaceConnectionRepository, WorkspaceConnectionRunRepository}; +use nvisy_postgres::types::{ConnectionId, Username}; use nvisy_postgres::{PgClient, PgConn}; use uuid::Uuid; @@ -67,7 +69,6 @@ async fn create_connection( let new_connection = NewWorkspaceConnection { workspace_id: workspace.id, account_id: auth_state.account_id, - slug: request.slug, name: request.name, provider: request.provider, encrypted_data, @@ -79,13 +80,17 @@ async fn create_connection( tracing::info!( target: TRACING_TARGET, - connection_slug = %connection.slug, + connection_id = %ConnectionId::from_uuid(connection.id), provider = %connection.provider, "Connection created", ); - let (connection, creator_username) = - find_connection(&mut conn, workspace.id, connection.slug.as_str()).await?; + let (connection, creator_username, last_synced) = find_connection( + &mut conn, + workspace.id, + ConnectionId::from_uuid(connection.id), + ) + .await?; Ok(( StatusCode::CREATED, @@ -93,6 +98,7 @@ async fn create_connection( connection, workspace.slug, creator_username, + last_synced, )), )) } @@ -144,6 +150,15 @@ async fn list_connections( ) .await?; + // One grouped query resolves last-synced for the whole page (not per row). + let ids: Vec = page.items.iter().map(|(c, _)| c.id).collect(); + let last_synced: HashMap = conn + .last_successful_sync_at(&ids) + .await? + .into_iter() + .map(|(id, ts)| (id, ts.into())) + .collect(); + tracing::debug!( target: TRACING_TARGET, connection_count = page.items.len(), @@ -155,7 +170,8 @@ async fn list_connections( Json(ConnectionsPage::from_cursor_page( page, |(connection, creator_username)| { - Connection::from_model(connection, workspace.slug.clone(), creator_username) + let synced = last_synced.get(&connection.id).copied(); + Connection::from_model(connection, workspace.slug.clone(), creator_username, synced) }, )), )) @@ -181,7 +197,7 @@ fn list_connections_docs(op: TransformOperation) -> TransformOperation { fields( account_id = %auth_state.account_id, workspace_id = %workspace.id, - connection_slug = %path_params.connection_slug, + connection_id = %path_params.connection_id, ) )] async fn read_connection( @@ -198,8 +214,8 @@ async fn read_connection( .authorize_workspace(&mut conn, workspace.id, Permission::ViewConnections) .await?; - let (connection, creator_username) = - find_connection(&mut conn, workspace.id, &path_params.connection_slug).await?; + let (connection, creator_username, last_synced) = + find_connection(&mut conn, workspace.id, path_params.connection_id).await?; tracing::debug!(target: TRACING_TARGET, "Workspace connection read"); @@ -209,6 +225,7 @@ async fn read_connection( connection, workspace.slug, creator_username, + last_synced, )), )) } @@ -230,7 +247,7 @@ fn read_connection_docs(op: TransformOperation) -> TransformOperation { fields( account_id = %auth_state.account_id, workspace_id = %workspace.id, - connection_slug = %path_params.connection_slug, + connection_id = %path_params.connection_id, ) )] async fn update_connection( @@ -249,8 +266,8 @@ async fn update_connection( .authorize_workspace(&mut conn, workspace.id, Permission::ManageConnections) .await?; - let (existing, _) = - find_connection(&mut conn, workspace.id, &path_params.connection_slug).await?; + let (existing, _, _) = + find_connection(&mut conn, workspace.id, path_params.connection_id).await?; let encrypted_data = request .data @@ -266,8 +283,8 @@ async fn update_connection( conn.update_workspace_connection(existing.id, update_data) .await?; - let (connection, creator_username) = - find_connection(&mut conn, workspace.id, &path_params.connection_slug).await?; + let (connection, creator_username, last_synced) = + find_connection(&mut conn, workspace.id, path_params.connection_id).await?; tracing::info!(target: TRACING_TARGET, "Connection updated"); @@ -277,6 +294,7 @@ async fn update_connection( connection, workspace.slug, creator_username, + last_synced, )), )) } @@ -299,7 +317,7 @@ fn update_connection_docs(op: TransformOperation) -> TransformOperation { fields( account_id = %auth_state.account_id, workspace_id = %workspace.id, - connection_slug = %path_params.connection_slug, + connection_id = %path_params.connection_id, ) )] async fn delete_connection( @@ -316,8 +334,8 @@ async fn delete_connection( .authorize_workspace(&mut conn, workspace.id, Permission::ManageConnections) .await?; - let (existing, _) = - find_connection(&mut conn, workspace.id, &path_params.connection_slug).await?; + let (existing, _, _) = + find_connection(&mut conn, workspace.id, path_params.connection_id).await?; conn.delete_workspace_connection(existing.id).await?; @@ -335,16 +353,24 @@ fn delete_connection_docs(op: TransformOperation) -> TransformOperation { .response::<404, Json>() } -/// Finds a connection within a workspace by slug, with its creator's handle, or +/// Finds a connection within a workspace by id, with its creator's handle, or /// returns a NotFound error. async fn find_connection( conn: &mut PgConn, workspace_id: Uuid, - connection_slug: &str, -) -> Result<(WorkspaceConnection, Username)> { - conn.find_connection_in_workspace_by_slug(workspace_id, connection_slug) + connection_id: ConnectionId, +) -> Result<(WorkspaceConnection, Username, Option)> { + let (connection, creator_username) = conn + .find_connection_in_workspace_with_creator(workspace_id, connection_id.as_uuid()) + .await? + .ok_or_else(|| Error::not_found("connection"))?; + let last_synced = conn + .last_successful_sync_at(&[connection.id]) .await? - .ok_or_else(|| Error::not_found("connection")) + .into_iter() + .next() + .map(|(_, ts)| ts.into()); + Ok((connection, creator_username, last_synced)) } /// Returns routes for workspace connection management. @@ -358,7 +384,7 @@ pub fn routes() -> ApiRouter { .get_with(list_connections, list_connections_docs), ) .api_route( - "/workspaces/{workspaceSlug}/connections/{connectionSlug}/", + "/workspaces/{workspaceSlug}/connections/{connectionId}/", get_with(read_connection, read_connection_docs) .put_with(update_connection, update_connection_docs) .delete_with(delete_connection, delete_connection_docs), diff --git a/crates/nvisy-server/src/handler/error/pg_pipeline.rs b/crates/nvisy-server/src/handler/error/pg_pipeline.rs index b0f6cc8..8baf47f 100644 --- a/crates/nvisy-server/src/handler/error/pg_pipeline.rs +++ b/crates/nvisy-server/src/handler/error/pg_pipeline.rs @@ -59,12 +59,6 @@ impl From for Error<'static> { WorkspacePipelineRunConstraints::IdempotencyKeyLength => { ErrorKind::BadRequest.with_message("Idempotency key must be 1 to 255 characters") } - WorkspacePipelineRunConstraints::RunNumberPositive => { - ErrorKind::BadRequest.with_message("Run number must be greater than zero") - } - WorkspacePipelineRunConstraints::RunNumberUnique => { - ErrorKind::Conflict.with_message("A run with this number already exists") - } WorkspacePipelineRunConstraints::CompletedAfterStarted => { ErrorKind::InternalServerError.into_error() } @@ -124,16 +118,11 @@ impl From for Error<'static> { } WorkspaceConnectionConstraints::MetadataSize => ErrorKind::BadRequest .with_message("Connection metadata size exceeds maximum limit"), - WorkspaceConnectionConstraints::SlugLength => ErrorKind::BadRequest - .with_message("Connection slug must be between 3 and 32 characters long"), - WorkspaceConnectionConstraints::SlugFormat => ErrorKind::BadRequest.with_message( - "Connection slug must be lowercase alphanumeric with single internal dashes", - ), - WorkspaceConnectionConstraints::SlugUnique => { - ErrorKind::Conflict.with_message("A connection with this slug already exists") - } WorkspaceConnectionConstraints::WorkspaceIdIdUnique => ErrorKind::Conflict .with_message("A connection with this identifier already exists"), + WorkspaceConnectionConstraints::NameUnique => { + ErrorKind::Conflict.with_message("A connection with this name already exists") + } WorkspaceConnectionConstraints::UpdatedAfterCreated | WorkspaceConnectionConstraints::DeletedAfterCreated => { ErrorKind::InternalServerError.into_error() @@ -152,12 +141,6 @@ impl From for Error<'static> { WorkspaceConnectionRunConstraints::MetadataSize => { ErrorKind::BadRequest.with_message("Sync run metadata size exceeds maximum limit") } - WorkspaceConnectionRunConstraints::RunNumberPositive => { - ErrorKind::BadRequest.with_message("Run number must be greater than zero") - } - WorkspaceConnectionRunConstraints::RunNumberUnique => { - ErrorKind::Conflict.with_message("A run with this number already exists") - } WorkspaceConnectionRunConstraints::RecordsSyncedNonNegative | WorkspaceConnectionRunConstraints::CompletedAfterStarted => { ErrorKind::InternalServerError.into_error() diff --git a/crates/nvisy-server/src/handler/error/pg_workspace.rs b/crates/nvisy-server/src/handler/error/pg_workspace.rs index 9004f51..19483ee 100644 --- a/crates/nvisy-server/src/handler/error/pg_workspace.rs +++ b/crates/nvisy-server/src/handler/error/pg_workspace.rs @@ -96,14 +96,6 @@ impl From for Error<'static> { impl From for Error<'static> { fn from(c: WorkspaceWebhookConstraints) -> Self { let error = match c { - WorkspaceWebhookConstraints::SlugLength => ErrorKind::BadRequest - .with_message("Webhook slug must be between 3 and 32 characters long"), - WorkspaceWebhookConstraints::SlugFormat => ErrorKind::BadRequest.with_message( - "Webhook slug must be lowercase alphanumeric with single internal dashes", - ), - WorkspaceWebhookConstraints::SlugUnique => { - ErrorKind::Conflict.with_message("A webhook with this slug already exists") - } WorkspaceWebhookConstraints::DisplayNameLength => ErrorKind::BadRequest .with_message("Webhook name must be between 3 and 64 characters long"), WorkspaceWebhookConstraints::DescriptionLength => { diff --git a/crates/nvisy-server/src/handler/mod.rs b/crates/nvisy-server/src/handler/mod.rs index 6795899..9a1fb18 100644 --- a/crates/nvisy-server/src/handler/mod.rs +++ b/crates/nvisy-server/src/handler/mod.rs @@ -5,7 +5,6 @@ mod accounts; mod authentication; -mod connection_runs; mod connections; mod contexts; mod error; @@ -14,11 +13,11 @@ mod invites; mod members; mod monitors; mod notifications; -mod pipeline_runs; mod pipelines; mod policies; pub mod request; pub mod response; +mod runs; mod tokens; mod utility; mod webhooks; @@ -66,9 +65,6 @@ fn private_routes( if is_included(BuiltinModule::Connections) { router = router.merge(connections::routes()); } - if is_included(BuiltinModule::ConnectionRuns) { - router = router.merge(connection_runs::routes()); - } if is_included(BuiltinModule::Contexts) { router = router.merge(contexts::routes()); } @@ -88,7 +84,7 @@ fn private_routes( router = router.merge(pipelines::routes()); } if is_included(BuiltinModule::PipelineRuns) { - router = router.merge(pipeline_runs::routes()); + router = router.merge(runs::routes()); } if is_included(BuiltinModule::Policies) { router = router.merge(policies::routes()); diff --git a/crates/nvisy-server/src/handler/request/connection_runs.rs b/crates/nvisy-server/src/handler/request/connection_runs.rs deleted file mode 100644 index 4c2e326..0000000 --- a/crates/nvisy-server/src/handler/request/connection_runs.rs +++ /dev/null @@ -1,29 +0,0 @@ -//! Connection sync run request types. - -use nvisy_postgres::types::SyncStatus; -use schemars::JsonSchema; -use serde::{Deserialize, Serialize}; - -/// Path parameters for connection sync run operations. -/// -/// The workspace is resolved separately from the `{workspaceSlug}` segment by -/// the [`WorkspaceContext`] extractor. -/// -/// [`WorkspaceContext`]: crate::extract::WorkspaceContext -#[must_use] -#[derive(Debug, Serialize, Deserialize, JsonSchema)] -#[serde(rename_all = "camelCase")] -pub struct ConnectionRunPathParams { - /// URL slug of the connection the run belongs to. - pub connection_slug: String, - /// Per-connection sequential run number. - pub run_number: i32, -} - -/// Query parameters for listing connection sync runs. -#[derive(Debug, Clone, Default, Serialize, Deserialize, JsonSchema)] -#[serde(rename_all = "camelCase")] -pub struct ConnectionRunsQuery { - /// Filter by run status. - pub status: Option, -} diff --git a/crates/nvisy-server/src/handler/request/connections.rs b/crates/nvisy-server/src/handler/request/connections.rs index 928d89d..74ad107 100644 --- a/crates/nvisy-server/src/handler/request/connections.rs +++ b/crates/nvisy-server/src/handler/request/connections.rs @@ -1,6 +1,6 @@ //! Connection request types. -use nvisy_postgres::types::Slug; +use nvisy_postgres::types::ConnectionId; use schemars::JsonSchema; use serde::{Deserialize, Serialize}; use validator::Validate; @@ -15,8 +15,8 @@ use validator::Validate; #[derive(Debug, Serialize, Deserialize, JsonSchema)] #[serde(rename_all = "camelCase")] pub struct ConnectionPathParams { - /// URL slug of the connection, unique within its workspace. - pub connection_slug: String, + /// Opaque identifier of the connection. + pub connection_id: ConnectionId, } /// Request payload for creating a new workspace connection. @@ -26,8 +26,6 @@ pub struct CreateConnection { /// Human-readable connection name. #[validate(length(min = 1, max = 255))] pub name: String, - /// URL slug, unique within the workspace and immutable after creation. - pub slug: Slug, /// Provider type (e.g., "openai", "postgres", "s3"). #[validate(length(min = 1, max = 64))] pub provider: String, diff --git a/crates/nvisy-server/src/handler/request/mod.rs b/crates/nvisy-server/src/handler/request/mod.rs index 874a15a..34511ba 100644 --- a/crates/nvisy-server/src/handler/request/mod.rs +++ b/crates/nvisy-server/src/handler/request/mod.rs @@ -2,7 +2,6 @@ mod accounts; mod authentications; -mod connection_runs; mod connections; mod contexts; mod files; @@ -21,7 +20,6 @@ mod workspaces; pub use accounts::*; pub use authentications::*; -pub use connection_runs::*; pub use connections::*; pub use contexts::*; pub use files::*; diff --git a/crates/nvisy-server/src/handler/request/paths.rs b/crates/nvisy-server/src/handler/request/paths.rs index b7e9925..0ef1d6b 100644 --- a/crates/nvisy-server/src/handler/request/paths.rs +++ b/crates/nvisy-server/src/handler/request/paths.rs @@ -1,6 +1,6 @@ //! Path parameter types for HTTP handlers. -use nvisy_postgres::types::Username; +use nvisy_postgres::types::{RunId, Username, WebhookId}; use schemars::JsonSchema; use serde::{Deserialize, Serialize}; use uuid::Uuid; @@ -46,8 +46,8 @@ pub struct WorkspaceFilePathParams { #[derive(Debug, Serialize, Deserialize, JsonSchema)] #[serde(rename_all = "camelCase")] pub struct WebhookPathParams { - /// URL slug of the webhook, unique within its workspace. - pub webhook_slug: String, + /// Opaque identifier of the webhook. + pub webhook_id: WebhookId, } /// Path parameters for API token operations. @@ -88,8 +88,6 @@ pub struct PipelinePathParams { #[derive(Debug, Serialize, Deserialize, JsonSchema)] #[serde(rename_all = "camelCase")] pub struct PipelineRunPathParams { - /// URL slug of the pipeline the run belongs to. - pub pipeline_slug: String, - /// Per-pipeline sequential run number. - pub run_number: i32, + /// Opaque identifier of the run. + pub run_id: RunId, } diff --git a/crates/nvisy-server/src/handler/request/webhooks.rs b/crates/nvisy-server/src/handler/request/webhooks.rs index a95297e..383d15c 100644 --- a/crates/nvisy-server/src/handler/request/webhooks.rs +++ b/crates/nvisy-server/src/handler/request/webhooks.rs @@ -8,7 +8,7 @@ use std::collections::HashMap; use nvisy_postgres::model::{ NewWorkspaceWebhook, UpdateWorkspaceWebhook as UpdateWorkspaceWebhookModel, }; -use nvisy_postgres::types::{Slug, WebhookEvent, WebhookStatus}; +use nvisy_postgres::types::{WebhookEvent, WebhookStatus}; use schemars::JsonSchema; use serde::{Deserialize, Serialize}; use uuid::Uuid; @@ -19,8 +19,6 @@ use validator::Validate; #[derive(Debug, Serialize, Deserialize, JsonSchema, Validate)] #[serde(rename_all = "camelCase")] pub struct CreateWebhook { - /// URL slug, unique within the workspace and immutable after creation. - pub slug: Slug, /// Human-readable name for the webhook (1-100 characters). #[validate(length(min = 1, max = 100))] pub display_name: String, @@ -62,7 +60,6 @@ impl CreateWebhook { NewWorkspaceWebhook { workspace_id, - slug: self.slug, display_name: self.display_name, description: self.description, url: self.url, diff --git a/crates/nvisy-server/src/handler/response/connection_runs.rs b/crates/nvisy-server/src/handler/response/connection_runs.rs deleted file mode 100644 index aa6cc11..0000000 --- a/crates/nvisy-server/src/handler/response/connection_runs.rs +++ /dev/null @@ -1,71 +0,0 @@ -//! Connection sync run response types. - -use jiff::Timestamp; -use nvisy_postgres::model::WorkspaceConnectionRun as ConnectionRunModel; -use nvisy_postgres::types::{Slug, SyncStatus, SyncTriggerType, Username}; -use schemars::JsonSchema; -use serde::{Deserialize, Serialize}; - -use super::Page; - -/// Response type for a connection sync run. -/// -/// A run is addressed as `(connection slug, run number)`, so it carries no -/// surrogate id of its own. -#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] -#[serde(rename_all = "camelCase")] -pub struct ConnectionRun { - /// Sequential run number within the connection (the run's identity). - pub run_number: i32, - /// Slug of the connection this run belongs to. - pub connection_slug: Slug, - /// Slug of the workspace this run belongs to. - pub workspace_slug: Slug, - /// Handle of the account that triggered the run, if any. - #[serde(skip_serializing_if = "Option::is_none")] - pub trigger_username: Option, - /// How the run was triggered. - pub trigger_type: SyncTriggerType, - /// Current run status. - pub status: SyncStatus, - /// Number of records processed by the run. - pub records_synced: i64, - /// Failure detail when the run failed. - #[serde(skip_serializing_if = "Option::is_none")] - pub error_message: Option, - /// Non-encrypted metadata for filtering/display. - pub metadata: serde_json::Value, - /// When the run started. - pub started_at: Timestamp, - /// When the run finished. - #[serde(skip_serializing_if = "Option::is_none")] - pub completed_at: Option, -} - -/// Paginated response for connection sync runs. -pub type ConnectionRunsPage = Page; - -impl ConnectionRun { - /// Creates a connection run response from the database model, the slugs of - /// its owning connection and workspace, and the triggering account's handle. - pub fn from_model( - run: ConnectionRunModel, - connection_slug: Slug, - workspace_slug: Slug, - trigger_username: Option, - ) -> Self { - Self { - run_number: run.run_number, - connection_slug, - workspace_slug, - trigger_username, - trigger_type: run.trigger_type, - status: run.status, - records_synced: run.records_synced, - error_message: run.error_message, - metadata: run.metadata, - started_at: run.started_at.into(), - completed_at: run.completed_at.map(Into::into), - } - } -} diff --git a/crates/nvisy-server/src/handler/response/connections.rs b/crates/nvisy-server/src/handler/response/connections.rs index d2a19d3..7cb22aa 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; -use nvisy_postgres::types::{Slug, Username}; +use nvisy_postgres::types::{ConnectionId, Slug, Username}; use schemars::JsonSchema; use serde::{Deserialize, Serialize}; @@ -15,8 +15,8 @@ use super::Page; #[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] #[serde(rename_all = "camelCase")] pub struct Connection { - /// URL slug of the connection, unique within its workspace. - pub slug: Slug, + /// Opaque identifier of the connection. + pub id: ConnectionId, /// Slug of the workspace this connection belongs to. pub workspace_slug: Slug, /// Handle of the account that created this connection. @@ -25,6 +25,9 @@ pub struct Connection { pub name: String, /// Provider type (e.g., "openai", "postgres", "s3"). pub provider: String, + /// When the connection last synced successfully, if ever. + #[serde(skip_serializing_if = "Option::is_none")] + pub last_synced: Option, /// When the connection was created. pub created_at: Timestamp, /// When the connection was last updated. @@ -40,13 +43,15 @@ impl Connection { connection: WorkspaceConnection, workspace_slug: Slug, creator_username: Username, + last_synced: Option, ) -> Self { Self { - slug: connection.slug, + id: ConnectionId::from_uuid(connection.id), workspace_slug, creator_username, name: connection.name, provider: connection.provider, + last_synced, created_at: connection.created_at.into(), updated_at: connection.updated_at.into(), } diff --git a/crates/nvisy-server/src/handler/response/mod.rs b/crates/nvisy-server/src/handler/response/mod.rs index 93768ce..8f8b360 100644 --- a/crates/nvisy-server/src/handler/response/mod.rs +++ b/crates/nvisy-server/src/handler/response/mod.rs @@ -8,7 +8,6 @@ mod accounts; mod activities; mod artifacts; mod authentications; -mod connection_runs; mod connections; mod contexts; mod errors; @@ -17,9 +16,9 @@ mod invites; mod members; mod monitors; mod notifications; -mod pipeline_runs; mod pipelines; mod policies; +mod runs; mod tokens; mod webhooks; mod workspaces; @@ -28,7 +27,6 @@ pub use accounts::*; pub use activities::*; pub use artifacts::*; pub use authentications::*; -pub use connection_runs::*; pub use connections::*; pub use contexts::*; pub use errors::*; @@ -37,9 +35,9 @@ pub use invites::*; pub use members::*; pub use monitors::*; pub use notifications::*; -pub use pipeline_runs::*; pub use pipelines::*; pub use policies::*; +pub use runs::*; pub use tokens::*; pub use webhooks::*; pub use workspaces::*; diff --git a/crates/nvisy-server/src/handler/response/pipeline_runs.rs b/crates/nvisy-server/src/handler/response/runs.rs similarity index 89% rename from crates/nvisy-server/src/handler/response/pipeline_runs.rs rename to crates/nvisy-server/src/handler/response/runs.rs index 7307f87..1ec63b1 100644 --- a/crates/nvisy-server/src/handler/response/pipeline_runs.rs +++ b/crates/nvisy-server/src/handler/response/runs.rs @@ -2,7 +2,7 @@ use jiff::Timestamp; use nvisy_postgres::model::WorkspacePipelineRun as PipelineRunModel; -use nvisy_postgres::types::{PipelineRunStatus, PipelineTriggerType, Slug, Username}; +use nvisy_postgres::types::{PipelineRunStatus, PipelineTriggerType, RunId, Slug, Username}; use schemars::JsonSchema; use serde::{Deserialize, Serialize}; use uuid::Uuid; @@ -11,13 +11,13 @@ use super::Page; /// Response type for a pipeline run. /// -/// A run is addressed as `(pipeline slug, run number)`, so it carries no -/// surrogate id of its own. +/// A run is addressed by its own opaque id; the owning pipeline and workspace +/// slugs are carried for context. #[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] #[serde(rename_all = "camelCase")] pub struct PipelineRun { - /// Sequential run number within the pipeline (the run's identity). - pub run_number: i32, + /// Opaque identifier of the run. + pub id: RunId, /// Slug of the pipeline this run belongs to. pub pipeline_slug: Slug, /// Slug of the workspace this run belongs to. @@ -56,7 +56,7 @@ impl PipelineRun { trigger_username: Option, ) -> Self { Self { - run_number: run.run_number, + id: RunId::from_uuid(run.id), pipeline_slug, workspace_slug, file_id: run.file_id, diff --git a/crates/nvisy-server/src/handler/response/webhooks.rs b/crates/nvisy-server/src/handler/response/webhooks.rs index afe2a99..326a124 100644 --- a/crates/nvisy-server/src/handler/response/webhooks.rs +++ b/crates/nvisy-server/src/handler/response/webhooks.rs @@ -4,7 +4,7 @@ use std::collections::HashMap; use jiff::Timestamp; use nvisy_postgres::model; -use nvisy_postgres::types::{Slug, Username, WebhookEvent, WebhookStatus}; +use nvisy_postgres::types::{Slug, Username, WebhookEvent, WebhookId, WebhookStatus}; use schemars::JsonSchema; use serde::{Deserialize, Serialize}; @@ -15,8 +15,8 @@ use super::Page; #[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] #[serde(rename_all = "camelCase")] pub struct Webhook { - /// URL slug of the webhook, unique within its workspace. - pub slug: Slug, + /// Opaque identifier of the webhook. + pub id: WebhookId, /// Slug of the workspace this webhook belongs to. pub workspace_slug: Slug, /// Human-readable name for the webhook. @@ -52,7 +52,7 @@ impl Webhook { let headers = webhook.parsed_headers(); Self { - slug: webhook.slug, + id: WebhookId::from_uuid(webhook.id), workspace_slug, display_name: webhook.display_name, description: webhook.description, diff --git a/crates/nvisy-server/src/handler/pipeline_runs.rs b/crates/nvisy-server/src/handler/runs.rs similarity index 95% rename from crates/nvisy-server/src/handler/pipeline_runs.rs rename to crates/nvisy-server/src/handler/runs.rs index b910707..5a84471 100644 --- a/crates/nvisy-server/src/handler/pipeline_runs.rs +++ b/crates/nvisy-server/src/handler/runs.rs @@ -155,7 +155,7 @@ async fn create_pipeline_run( ) .await?; - tracing::info!(target: TRACING_TARGET, run_number = run.run_number, "Pipeline run analyzed"); + tracing::info!(target: TRACING_TARGET, run_id = %run.id, "Pipeline run analyzed"); let trigger_username = resolve_trigger_username(&mut conn, run.account_id).await?; @@ -314,8 +314,7 @@ fn list_workspace_runs_docs(op: TransformOperation) -> TransformOperation { fields( account_id = %auth_state.account_id, workspace_id = %workspace.id, - pipeline_slug = %path_params.pipeline_slug, - run_number = path_params.run_number, + run_id = %path_params.run_id, ) )] async fn get_pipeline_run( @@ -332,13 +331,8 @@ async fn get_pipeline_run( .authorize_workspace(&mut conn, workspace.id, Permission::ViewPipelines) .await?; - let (pipeline, run, trigger_username) = find_pipeline_run( - &mut conn, - workspace.id, - &path_params.pipeline_slug, - path_params.run_number, - ) - .await?; + let (pipeline, run, trigger_username) = + find_pipeline_run(&mut conn, workspace.id, path_params.run_id.as_uuid()).await?; tracing::debug!(target: TRACING_TARGET, "Pipeline run retrieved"); @@ -371,8 +365,7 @@ fn get_pipeline_run_docs(op: TransformOperation) -> TransformOperation { fields( account_id = %auth_state.account_id, workspace_id = %workspace.id, - pipeline_slug = %path_params.pipeline_slug, - run_number = path_params.run_number, + run_id = %path_params.run_id, ) )] async fn get_pipeline_run_analysis( @@ -391,13 +384,8 @@ async fn get_pipeline_run_analysis( .authorize_workspace(&mut conn, workspace.id, Permission::ViewPipelines) .await?; - let (_pipeline, run, _) = find_pipeline_run( - &mut conn, - workspace.id, - &path_params.pipeline_slug, - path_params.run_number, - ) - .await?; + let (_pipeline, run, _) = + find_pipeline_run(&mut conn, workspace.id, path_params.run_id.as_uuid()).await?; let analyzed = load_analyzed_document(&nats, &crypto, workspace.id, &run).await?; @@ -426,8 +414,7 @@ fn get_pipeline_run_analysis_docs(op: TransformOperation) -> TransformOperation fields( account_id = %auth_state.account_id, workspace_id = %workspace.id, - pipeline_slug = %path_params.pipeline_slug, - run_number = path_params.run_number, + run_id = %path_params.run_id, ) )] async fn redact_pipeline_run( @@ -447,13 +434,8 @@ async fn redact_pipeline_run( .authorize_workspace(&mut conn, workspace.id, Permission::RunPipelines) .await?; - let (pipeline, run, trigger_username) = find_pipeline_run( - &mut conn, - workspace.id, - &path_params.pipeline_slug, - path_params.run_number, - ) - .await?; + let (pipeline, run, trigger_username) = + find_pipeline_run(&mut conn, workspace.id, path_params.run_id.as_uuid()).await?; // A run can only be redacted once, after detection. if !run.is_analyzed() { @@ -502,7 +484,7 @@ async fn redact_pipeline_run( tracing::info!( target: TRACING_TARGET, - run_number = run.run_number, + run_id = %run.id, artifact_file_id = %artifact_file.id, "Pipeline run redacted" ); @@ -604,15 +586,16 @@ async fn resolve_trigger_username( .map(|account| account.username)) } +/// Resolves a run by its opaque id within a workspace, returning the run, its +/// owning pipeline (for the response's pipeline slug), and the triggering +/// account's handle. The lookup is workspace-scoped through the owning pipeline. async fn find_pipeline_run( conn: &mut PgConn, workspace_id: Uuid, - pipeline_slug: &str, - run_number: i32, + run_id: Uuid, ) -> Result<(WorkspacePipeline, WorkspacePipelineRun, Option)> { - let pipeline = find_pipeline(conn, workspace_id, pipeline_slug).await?; - let (run, trigger_username) = conn - .find_pipeline_run_by_number(pipeline.id, run_number) + let (run, pipeline, trigger_username) = conn + .find_workspace_run_by_id(workspace_id, run_id) .await? .ok_or_else(|| Error::not_found("pipeline_run"))?; Ok((pipeline, run, trigger_username)) @@ -635,15 +618,15 @@ pub fn routes() -> ApiRouter { .get_with(list_pipeline_runs, list_pipeline_runs_docs), ) .api_route( - "/workspaces/{workspaceSlug}/pipelines/{pipelineSlug}/runs/{runNumber}/", + "/workspaces/{workspaceSlug}/runs/{runId}/", get_with(get_pipeline_run, get_pipeline_run_docs), ) .api_route( - "/workspaces/{workspaceSlug}/pipelines/{pipelineSlug}/runs/{runNumber}/detections/", + "/workspaces/{workspaceSlug}/runs/{runId}/detections/", get_with(get_pipeline_run_analysis, get_pipeline_run_analysis_docs), ) .api_route( - "/workspaces/{workspaceSlug}/pipelines/{pipelineSlug}/runs/{runNumber}/redactions/", + "/workspaces/{workspaceSlug}/runs/{runId}/redactions/", post_with(redact_pipeline_run, redact_pipeline_run_docs), ) .with_path_items(|item| item.tag("Pipeline Runs")) diff --git a/crates/nvisy-server/src/handler/utility/custom_routes.rs b/crates/nvisy-server/src/handler/utility/custom_routes.rs index 1c091a6..1ee7e5a 100644 --- a/crates/nvisy-server/src/handler/utility/custom_routes.rs +++ b/crates/nvisy-server/src/handler/utility/custom_routes.rs @@ -24,8 +24,6 @@ pub enum BuiltinModule { Workspaces, /// Provider connections. Connections, - /// Connection sync runs. - ConnectionRuns, /// Detection contexts. Contexts, /// Workspace invitations. diff --git a/crates/nvisy-server/src/handler/webhooks.rs b/crates/nvisy-server/src/handler/webhooks.rs index 1ad50a4..874e7c6 100644 --- a/crates/nvisy-server/src/handler/webhooks.rs +++ b/crates/nvisy-server/src/handler/webhooks.rs @@ -69,12 +69,11 @@ async fn create_webhook( tracing::info!( target: TRACING_TARGET, - webhook_slug = %webhook.slug, + webhook_id = %webhook.id, "Webhook created", ); - let (webhook, creator_username) = - find_webhook(&mut conn, workspace.id, webhook.slug.as_str()).await?; + let (webhook, creator_username) = find_webhook(&mut conn, workspace.id, webhook.id).await?; // Return WebhookCreated which includes the secret (visible only once) Ok(( @@ -162,7 +161,7 @@ fn list_webhooks_docs(op: TransformOperation) -> TransformOperation { fields( account_id = %auth_state.account_id, workspace_id = %workspace.id, - webhook_slug = %path_params.webhook_slug, + webhook_id = %path_params.webhook_id, ) )] async fn read_webhook( @@ -180,7 +179,7 @@ async fn read_webhook( .await?; let (webhook, creator_username) = - find_webhook(&mut conn, workspace.id, &path_params.webhook_slug).await?; + find_webhook(&mut conn, workspace.id, path_params.webhook_id.as_uuid()).await?; tracing::debug!(target: TRACING_TARGET, "Workspace webhook read"); @@ -211,7 +210,7 @@ fn read_webhook_docs(op: TransformOperation) -> TransformOperation { fields( account_id = %auth_state.account_id, workspace_id = %workspace.id, - webhook_slug = %path_params.webhook_slug, + webhook_id = %path_params.webhook_id, ) )] async fn update_webhook( @@ -229,14 +228,15 @@ async fn update_webhook( .authorize_workspace(&mut conn, workspace.id, Permission::UpdateWebhooks) .await?; - let (existing, _) = find_webhook(&mut conn, workspace.id, &path_params.webhook_slug).await?; + let (existing, _) = + find_webhook(&mut conn, workspace.id, path_params.webhook_id.as_uuid()).await?; let update_data = request.into_model(existing.status); conn.update_workspace_webhook(existing.id, update_data) .await?; let (webhook, creator_username) = - find_webhook(&mut conn, workspace.id, &path_params.webhook_slug).await?; + find_webhook(&mut conn, workspace.id, path_params.webhook_id.as_uuid()).await?; tracing::info!(target: TRACING_TARGET, "Webhook updated"); @@ -268,7 +268,7 @@ fn update_webhook_docs(op: TransformOperation) -> TransformOperation { fields( account_id = %auth_state.account_id, workspace_id = %workspace.id, - webhook_slug = %path_params.webhook_slug, + webhook_id = %path_params.webhook_id, ) )] async fn delete_webhook( @@ -285,7 +285,8 @@ async fn delete_webhook( .authorize_workspace(&mut conn, workspace.id, Permission::DeleteWebhooks) .await?; - let (existing, _) = find_webhook(&mut conn, workspace.id, &path_params.webhook_slug).await?; + let (existing, _) = + find_webhook(&mut conn, workspace.id, path_params.webhook_id.as_uuid()).await?; conn.delete_workspace_webhook(existing.id).await?; @@ -312,7 +313,7 @@ fn delete_webhook_docs(op: TransformOperation) -> TransformOperation { fields( account_id = %auth_state.account_id, workspace_id = %workspace.id, - webhook_slug = %path_params.webhook_slug, + webhook_id = %path_params.webhook_id, ) )] async fn test_webhook( @@ -331,7 +332,8 @@ async fn test_webhook( .authorize_workspace(&mut conn, workspace.id, Permission::TestWebhooks) .await?; - let (webhook, _) = find_webhook(&mut conn, workspace.id, &path_params.webhook_slug).await?; + let (webhook, _) = + find_webhook(&mut conn, workspace.id, path_params.webhook_id.as_uuid()).await?; // Parse the webhook URL let url: Url = webhook.url.parse().map_err(|_| { @@ -370,14 +372,14 @@ fn test_webhook_docs(op: TransformOperation) -> TransformOperation { .response::<404, Json>() } -/// Finds a webhook within a workspace by slug, with its creator's handle, or +/// Finds a webhook within a workspace by id, with its creator's handle, or /// returns a NotFound error. async fn find_webhook( conn: &mut PgConn, workspace_id: Uuid, - webhook_slug: &str, + webhook_id: Uuid, ) -> Result<(WorkspaceWebhook, Username)> { - conn.find_webhook_in_workspace_by_slug(workspace_id, webhook_slug) + conn.find_webhook_in_workspace_with_creator(workspace_id, webhook_id) .await? .ok_or_else(|| Error::not_found("webhook")) } @@ -393,13 +395,13 @@ pub fn routes() -> ApiRouter { .get_with(list_webhooks, list_webhooks_docs), ) .api_route( - "/workspaces/{workspaceSlug}/webhooks/{webhookSlug}/", + "/workspaces/{workspaceSlug}/webhooks/{webhookId}/", get_with(read_webhook, read_webhook_docs) .put_with(update_webhook, update_webhook_docs) .delete_with(delete_webhook, delete_webhook_docs), ) .api_route( - "/workspaces/{workspaceSlug}/webhooks/{webhookSlug}/test/", + "/workspaces/{workspaceSlug}/webhooks/{webhookId}/test/", post_with(test_webhook, test_webhook_docs), ) .with_path_items(|item| item.tag("Webhooks")) diff --git a/migrations/2025-05-21-222842_webhooks/up.sql b/migrations/2025-05-21-222842_webhooks/up.sql index 5771893..35794e9 100644 --- a/migrations/2025-05-21-222842_webhooks/up.sql +++ b/migrations/2025-05-21-222842_webhooks/up.sql @@ -45,13 +45,6 @@ CREATE TABLE workspace_webhooks ( -- Composite key target for workspace-scoped access and foreign keys. CONSTRAINT workspace_webhooks_workspace_id_id_key UNIQUE (workspace_id, id), - -- URL identity, unique within the workspace: lowercase alphanumeric with - -- single internal dashes, 3-32 characters. - slug TEXT NOT NULL, - CONSTRAINT workspace_webhooks_workspace_id_slug_key UNIQUE (workspace_id, slug), - CONSTRAINT workspace_webhooks_slug_length CHECK (length(slug) BETWEEN 3 AND 32), - CONSTRAINT workspace_webhooks_slug_format CHECK (slug ~ '^[a-z0-9]+(-[a-z0-9]+)*$'), - -- Webhook details display_name TEXT NOT NULL, description TEXT NOT NULL DEFAULT '', diff --git a/migrations/2026-01-19-045013_connections/down.sql b/migrations/2026-01-19-045013_connections/down.sql index 3046e92..442d62e 100644 --- a/migrations/2026-01-19-045013_connections/down.sql +++ b/migrations/2026-01-19-045013_connections/down.sql @@ -4,9 +4,6 @@ DROP VIEW IF EXISTS workspace_connection_sync_state; DROP TABLE IF EXISTS workspace_connection_runs; --- Drop the run_number trigger function (its trigger dropped with the table above). -DROP FUNCTION IF EXISTS set_workspace_connection_run_number(); - DROP TABLE IF EXISTS workspace_connections; DROP TYPE IF EXISTS SYNC_TRIGGER_TYPE; diff --git a/migrations/2026-01-19-045013_connections/up.sql b/migrations/2026-01-19-045013_connections/up.sql index e10b547..928c623 100644 --- a/migrations/2026-01-19-045013_connections/up.sql +++ b/migrations/2026-01-19-045013_connections/up.sql @@ -34,13 +34,6 @@ CREATE TABLE workspace_connections ( -- Composite key target for workspace-scoped access and foreign keys. CONSTRAINT workspace_connections_workspace_id_id_key UNIQUE (workspace_id, id), - -- URL identity, unique within the workspace: lowercase alphanumeric with - -- single internal dashes, 3-32 characters. - slug TEXT NOT NULL, - CONSTRAINT workspace_connections_workspace_id_slug_key UNIQUE (workspace_id, slug), - CONSTRAINT workspace_connections_slug_length CHECK (length(slug) BETWEEN 3 AND 32), - CONSTRAINT workspace_connections_slug_format CHECK (slug ~ '^[a-z0-9]+(-[a-z0-9]+)*$'), - -- Core attributes name TEXT NOT NULL, provider TEXT NOT NULL, @@ -123,13 +116,6 @@ CREATE TABLE workspace_connection_runs ( trigger_type SYNC_TRIGGER_TYPE NOT NULL DEFAULT 'manual', status SYNC_STATUS NOT NULL DEFAULT 'running', - -- Human-facing sequence number within the connection ("run #47"). Assigned - -- at insert; display only, never used to address the run (the UUID is). - run_number INTEGER NOT NULL, - - CONSTRAINT workspace_connection_runs_run_number_positive CHECK (run_number >= 1), - CONSTRAINT workspace_connection_runs_connection_id_run_number_key UNIQUE (connection_id, run_number), - -- Number of records processed by this run. Resumption state (cursor, -- offset) is not stored here; it lives in the connection's encrypted -- context, which each run reads and advances. @@ -154,29 +140,6 @@ CREATE TABLE workspace_connection_runs ( CONSTRAINT workspace_connection_runs_completed_after_started CHECK (completed_at IS NULL OR completed_at >= started_at) ); --- Assign a gap-free, per-connection run_number on insert. Locking the parent --- connection row serializes concurrent inserts for the same connection --- (different connections proceed in parallel), so the sequence is contiguous --- and race-free. -CREATE OR REPLACE FUNCTION set_workspace_connection_run_number() -RETURNS TRIGGER AS $$ -BEGIN - PERFORM 1 FROM workspace_connections WHERE id = NEW.connection_id FOR UPDATE; - SELECT COALESCE(MAX(run_number), 0) + 1 INTO NEW.run_number - FROM workspace_connection_runs - WHERE connection_id = NEW.connection_id; - RETURN NEW; -END; -$$ LANGUAGE plpgsql; - -CREATE TRIGGER workspace_connection_runs_set_run_number_trigger - BEFORE INSERT ON workspace_connection_runs - FOR EACH ROW - EXECUTE FUNCTION set_workspace_connection_run_number(); - -COMMENT ON FUNCTION set_workspace_connection_run_number() IS - 'Assigns a gap-free per-connection run_number, serialized by locking the connection row.'; - -- Indexes CREATE INDEX workspace_connection_runs_connection_idx ON workspace_connection_runs (connection_id, started_at DESC); diff --git a/migrations/2026-01-19-045016_pipelines/down.sql b/migrations/2026-01-19-045016_pipelines/down.sql index 2a66a93..168b92b 100644 --- a/migrations/2026-01-19-045016_pipelines/down.sql +++ b/migrations/2026-01-19-045016_pipelines/down.sql @@ -15,9 +15,6 @@ DROP VIEW IF EXISTS active_workspace_pipeline_runs; -- Pipeline runs DROP TABLE IF EXISTS workspace_pipeline_runs; --- Drop the run_number trigger function (its trigger dropped with the table above). -DROP FUNCTION IF EXISTS set_workspace_pipeline_run_number(); - -- Workspace pipelines DROP TABLE IF EXISTS workspace_pipelines; diff --git a/migrations/2026-01-19-045016_pipelines/up.sql b/migrations/2026-01-19-045016_pipelines/up.sql index 1091402..03f0b1f 100644 --- a/migrations/2026-01-19-045016_pipelines/up.sql +++ b/migrations/2026-01-19-045016_pipelines/up.sql @@ -184,13 +184,6 @@ CREATE TABLE workspace_pipeline_runs ( trigger_type PIPELINE_TRIGGER_TYPE NOT NULL DEFAULT 'manual', status PIPELINE_RUN_STATUS NOT NULL DEFAULT 'running', - -- Human-facing sequence number within the pipeline ("run #47"). Assigned at - -- insert; display only, never used to address the run (the UUID is). - run_number INTEGER NOT NULL, - - CONSTRAINT workspace_pipeline_runs_run_number_positive CHECK (run_number >= 1), - CONSTRAINT workspace_pipeline_runs_pipeline_id_run_number_key UNIQUE (pipeline_id, run_number), - -- Object-store key for the engine's AnalyzedDocument, encrypted and held in -- the intermediates bucket between the detect and redact calls. Null until -- analysis writes it; redact reads it back as the source of truth. @@ -216,28 +209,6 @@ CREATE TABLE workspace_pipeline_runs ( CONSTRAINT workspace_pipeline_runs_completed_after_started CHECK (completed_at IS NULL OR completed_at >= started_at) ); --- Assign a gap-free, per-pipeline run_number on insert. Locking the parent --- pipeline row serializes concurrent inserts for the same pipeline (different --- pipelines proceed in parallel), so the sequence is contiguous and race-free. -CREATE OR REPLACE FUNCTION set_workspace_pipeline_run_number() -RETURNS TRIGGER AS $$ -BEGIN - PERFORM 1 FROM workspace_pipelines WHERE id = NEW.pipeline_id FOR UPDATE; - SELECT COALESCE(MAX(run_number), 0) + 1 INTO NEW.run_number - FROM workspace_pipeline_runs - WHERE pipeline_id = NEW.pipeline_id; - RETURN NEW; -END; -$$ LANGUAGE plpgsql; - -CREATE TRIGGER workspace_pipeline_runs_set_run_number_trigger - BEFORE INSERT ON workspace_pipeline_runs - FOR EACH ROW - EXECUTE FUNCTION set_workspace_pipeline_run_number(); - -COMMENT ON FUNCTION set_workspace_pipeline_run_number() IS - 'Assigns a gap-free per-pipeline run_number, serialized by locking the pipeline row.'; - -- Indexes CREATE INDEX workspace_pipeline_runs_pipeline_idx ON workspace_pipeline_runs (pipeline_id, started_at DESC);