refactor: group postgres adapters by bounded context

crates/adapters/postgres/src/processing/mod.rs

@@ -0,0 +1,532 @@
+use crate::db::PgPool;
+use async_trait::async_trait;
+use chrono::{DateTime, Utc};
+use domain::{
+    entities::{
+        BatchStatus, Job, JobBatch, JobStatus, JobType, PipelineStep, Plugin, PluginType,
+        ProcessingPipeline,
+    },
+    errors::DomainError,
+    ports::{JobBatchRepository, JobRepository, PipelineRepository, PluginRepository},
+    value_objects::{DateTimeStamp, MetadataValue, StructuredData, SystemId},
+};
+use uuid::Uuid;
+
+// ──────────────────────────────────────────────
+// Job
+// ──────────────────────────────────────────────
+
+#[derive(sqlx::FromRow)]
+struct JobRow {
+    job_id: Uuid,
+    job_type: String,
+    target_asset_id: Option<Uuid>,
+    batch_id: Option<Uuid>,
+    status: String,
+    priority: i32,
+    payload: serde_json::Value,
+    result_data: Option<serde_json::Value>,
+    retry_count: i32,
+    max_retries: i32,
+    created_at: DateTime<Utc>,
+    started_at: Option<DateTime<Utc>>,
+    completed_at: Option<DateTime<Utc>>,
+    error_message: Option<String>,
+}
+
+fn job_type_from_str(s: &str) -> JobType {
+    match s {
+        "scan_directory" => JobType::ScanDirectory,
+        "extract_metadata" => JobType::ExtractMetadata,
+        "generate_derivative" => JobType::GenerateDerivative,
+        "sync_sidecar" => JobType::SyncSidecar,
+        "detect_duplicates" => JobType::DetectDuplicates,
+        other => JobType::Custom(other.to_string()),
+    }
+}
+
+fn job_type_to_str(t: &JobType) -> String {
+    match t {
+        JobType::ScanDirectory => "scan_directory".to_string(),
+        JobType::ExtractMetadata => "extract_metadata".to_string(),
+        JobType::GenerateDerivative => "generate_derivative".to_string(),
+        JobType::SyncSidecar => "sync_sidecar".to_string(),
+        JobType::DetectDuplicates => "detect_duplicates".to_string(),
+        JobType::Custom(s) => s.clone(),
+    }
+}
+
+fn job_status_from_str(s: &str) -> JobStatus {
+    match s {
+        "queued" => JobStatus::Queued,
+        "processing" => JobStatus::Processing,
+        "completed" => JobStatus::Completed,
+        "failed" => JobStatus::Failed,
+        "cancelled" => JobStatus::Cancelled,
+        _ => JobStatus::Queued,
+    }
+}
+
+fn job_status_to_str(s: &JobStatus) -> &'static str {
+    match s {
+        JobStatus::Queued => "queued",
+        JobStatus::Processing => "processing",
+        JobStatus::Completed => "completed",
+        JobStatus::Failed => "failed",
+        JobStatus::Cancelled => "cancelled",
+    }
+}
+
+fn structured_from_json(v: serde_json::Value) -> StructuredData {
+    if let serde_json::Value::Object(map) = v {
+        let mut sd = StructuredData::new();
+        for (k, val) in map {
+            sd.insert(k, MetadataValue::from(val));
+        }
+        sd
+    } else {
+        StructuredData::new()
+    }
+}
+
+fn structured_to_json(sd: &StructuredData) -> serde_json::Value {
+    let map: serde_json::Map<String, serde_json::Value> = sd
+        .inner()
+        .iter()
+        .map(|(k, v)| (k.clone(), serde_json::Value::from(v)))
+        .collect();
+    serde_json::Value::Object(map)
+}
+
+impl From<JobRow> for Job {
+    fn from(r: JobRow) -> Self {
+        Self {
+            job_id: SystemId::from_uuid(r.job_id),
+            job_type: job_type_from_str(&r.job_type),
+            target_asset_id: r.target_asset_id.map(SystemId::from_uuid),
+            batch_id: r.batch_id.map(SystemId::from_uuid),
+            status: job_status_from_str(&r.status),
+            priority: r.priority as u32,
+            payload: structured_from_json(r.payload),
+            result_data: r.result_data.map(structured_from_json),
+            retry_count: r.retry_count as u32,
+            max_retries: r.max_retries as u32,
+            created_at: DateTimeStamp::from_datetime(r.created_at),
+            started_at: r.started_at.map(DateTimeStamp::from_datetime),
+            completed_at: r.completed_at.map(DateTimeStamp::from_datetime),
+            error_message: r.error_message,
+        }
+    }
+}
+
+pub struct PostgresJobRepository {
+    pool: PgPool,
+}
+
+impl PostgresJobRepository {
+    pub fn new(pool: PgPool) -> Self {
+        Self { pool }
+    }
+}
+
+#[async_trait]
+impl JobRepository for PostgresJobRepository {
+    async fn find_by_id(&self, id: &SystemId) -> Result<Option<Job>, DomainError> {
+        let row = sqlx::query_as::<_, JobRow>(
+            "SELECT job_id, job_type, target_asset_id, batch_id, status, priority,
+                    payload, result_data, retry_count, max_retries, created_at,
+                    started_at, completed_at, error_message
+             FROM jobs WHERE job_id = $1",
+        )
+        .bind(*id.as_uuid())
+        .fetch_optional(&self.pool)
+        .await
+        .map_err(|e| DomainError::Internal(e.to_string()))?;
+
+        Ok(row.map(Into::into))
+    }
+
+    async fn find_next_queued(&self) -> Result<Option<Job>, DomainError> {
+        let row = sqlx::query_as::<_, JobRow>(
+            "SELECT job_id, job_type, target_asset_id, batch_id, status, priority,
+                    payload, result_data, retry_count, max_retries, created_at,
+                    started_at, completed_at, error_message
+             FROM jobs WHERE status = 'queued'
+             ORDER BY priority DESC, created_at ASC
+             LIMIT 1",
+        )
+        .fetch_optional(&self.pool)
+        .await
+        .map_err(|e| DomainError::Internal(e.to_string()))?;
+
+        Ok(row.map(Into::into))
+    }
+
+    async fn find_by_batch(&self, batch_id: &SystemId) -> Result<Vec<Job>, DomainError> {
+        let rows = sqlx::query_as::<_, JobRow>(
+            "SELECT job_id, job_type, target_asset_id, batch_id, status, priority,
+                    payload, result_data, retry_count, max_retries, created_at,
+                    started_at, completed_at, error_message
+             FROM jobs WHERE batch_id = $1
+             ORDER BY created_at ASC",
+        )
+        .bind(*batch_id.as_uuid())
+        .fetch_all(&self.pool)
+        .await
+        .map_err(|e| DomainError::Internal(e.to_string()))?;
+
+        Ok(rows.into_iter().map(Into::into).collect())
+    }
+
+    async fn save(&self, job: &Job) -> Result<(), DomainError> {
+        sqlx::query(
+            "INSERT INTO jobs (job_id, job_type, target_asset_id, batch_id, status, priority,
+                    payload, result_data, retry_count, max_retries, created_at,
+                    started_at, completed_at, error_message)
+             VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14)
+             ON CONFLICT (job_id) DO UPDATE SET
+               status = EXCLUDED.status,
+               priority = EXCLUDED.priority,
+               payload = EXCLUDED.payload,
+               result_data = EXCLUDED.result_data,
+               retry_count = EXCLUDED.retry_count,
+               started_at = EXCLUDED.started_at,
+               completed_at = EXCLUDED.completed_at,
+               error_message = EXCLUDED.error_message",
+        )
+        .bind(*job.job_id.as_uuid())
+        .bind(job_type_to_str(&job.job_type))
+        .bind(job.target_asset_id.as_ref().map(|id| *id.as_uuid()))
+        .bind(job.batch_id.as_ref().map(|id| *id.as_uuid()))
+        .bind(job_status_to_str(&job.status))
+        .bind(job.priority as i32)
+        .bind(structured_to_json(&job.payload))
+        .bind(job.result_data.as_ref().map(structured_to_json))
+        .bind(job.retry_count as i32)
+        .bind(job.max_retries as i32)
+        .bind(job.created_at.as_datetime())
+        .bind(job.started_at.as_ref().map(|d| d.as_datetime()))
+        .bind(job.completed_at.as_ref().map(|d| d.as_datetime()))
+        .bind(&job.error_message)
+        .execute(&self.pool)
+        .await
+        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        Ok(())
+    }
+}
+
+// ──────────────────────────────────────────────
+// JobBatch
+// ──────────────────────────────────────────────
+
+#[derive(sqlx::FromRow)]
+struct BatchRow {
+    batch_id: Uuid,
+    batch_type: String,
+    total_jobs: i32,
+    completed_count: i32,
+    failed_count: i32,
+    status: String,
+}
+
+fn batch_status_from_str(s: &str) -> BatchStatus {
+    match s {
+        "in_progress" => BatchStatus::InProgress,
+        "completed_with_errors" => BatchStatus::CompletedWithErrors,
+        "completed" => BatchStatus::Completed,
+        "cancelled" => BatchStatus::Cancelled,
+        _ => BatchStatus::InProgress,
+    }
+}
+
+fn batch_status_to_str(s: &BatchStatus) -> &'static str {
+    match s {
+        BatchStatus::InProgress => "in_progress",
+        BatchStatus::CompletedWithErrors => "completed_with_errors",
+        BatchStatus::Completed => "completed",
+        BatchStatus::Cancelled => "cancelled",
+    }
+}
+
+impl From<BatchRow> for JobBatch {
+    fn from(r: BatchRow) -> Self {
+        Self {
+            batch_id: SystemId::from_uuid(r.batch_id),
+            batch_type: r.batch_type,
+            total_jobs: r.total_jobs as u32,
+            completed_count: r.completed_count as u32,
+            failed_count: r.failed_count as u32,
+            status: batch_status_from_str(&r.status),
+        }
+    }
+}
+
+pub struct PostgresJobBatchRepository {
+    pool: PgPool,
+}
+
+impl PostgresJobBatchRepository {
+    pub fn new(pool: PgPool) -> Self {
+        Self { pool }
+    }
+}
+
+#[async_trait]
+impl JobBatchRepository for PostgresJobBatchRepository {
+    async fn find_by_id(&self, id: &SystemId) -> Result<Option<JobBatch>, DomainError> {
+        let row = sqlx::query_as::<_, BatchRow>(
+            "SELECT batch_id, batch_type, total_jobs, completed_count, failed_count, status
+             FROM job_batches WHERE batch_id = $1",
+        )
+        .bind(*id.as_uuid())
+        .fetch_optional(&self.pool)
+        .await
+        .map_err(|e| DomainError::Internal(e.to_string()))?;
+
+        Ok(row.map(Into::into))
+    }
+
+    async fn save(&self, batch: &JobBatch) -> Result<(), DomainError> {
+        sqlx::query(
+            "INSERT INTO job_batches (batch_id, batch_type, total_jobs, completed_count, failed_count, status)
+             VALUES ($1, $2, $3, $4, $5, $6)
+             ON CONFLICT (batch_id) DO UPDATE SET
+               total_jobs = EXCLUDED.total_jobs,
+               completed_count = EXCLUDED.completed_count,
+               failed_count = EXCLUDED.failed_count,
+               status = EXCLUDED.status",
+        )
+        .bind(*batch.batch_id.as_uuid())
+        .bind(&batch.batch_type)
+        .bind(batch.total_jobs as i32)
+        .bind(batch.completed_count as i32)
+        .bind(batch.failed_count as i32)
+        .bind(batch_status_to_str(&batch.status))
+        .execute(&self.pool)
+        .await
+        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        Ok(())
+    }
+}
+
+// ──────────────────────────────────────────────
+// Plugin
+// ──────────────────────────────────────────────
+
+#[derive(sqlx::FromRow)]
+struct PluginRow {
+    plugin_id: Uuid,
+    name: String,
+    plugin_type: String,
+    is_enabled: bool,
+    configuration: serde_json::Value,
+}
+
+fn plugin_type_from_str(s: &str) -> PluginType {
+    match s {
+        "media_processor" => PluginType::MediaProcessor,
+        "scheduled_task" => PluginType::ScheduledTask,
+        "sidecar_writer" => PluginType::SidecarWriter,
+        _ => PluginType::MediaProcessor,
+    }
+}
+
+fn plugin_type_to_str(t: &PluginType) -> &'static str {
+    match t {
+        PluginType::MediaProcessor => "media_processor",
+        PluginType::ScheduledTask => "scheduled_task",
+        PluginType::SidecarWriter => "sidecar_writer",
+    }
+}
+
+impl From<PluginRow> for Plugin {
+    fn from(r: PluginRow) -> Self {
+        Self {
+            plugin_id: SystemId::from_uuid(r.plugin_id),
+            name: r.name,
+            plugin_type: plugin_type_from_str(&r.plugin_type),
+            is_enabled: r.is_enabled,
+            configuration: structured_from_json(r.configuration),
+        }
+    }
+}
+
+pub struct PostgresPluginRepository {
+    pool: PgPool,
+}
+
+impl PostgresPluginRepository {
+    pub fn new(pool: PgPool) -> Self {
+        Self { pool }
+    }
+}
+
+#[async_trait]
+impl PluginRepository for PostgresPluginRepository {
+    async fn find_by_id(&self, id: &SystemId) -> Result<Option<Plugin>, DomainError> {
+        let row = sqlx::query_as::<_, PluginRow>(
+            "SELECT plugin_id, name, plugin_type, is_enabled, configuration
+             FROM plugins WHERE plugin_id = $1",
+        )
+        .bind(*id.as_uuid())
+        .fetch_optional(&self.pool)
+        .await
+        .map_err(|e| DomainError::Internal(e.to_string()))?;
+
+        Ok(row.map(Into::into))
+    }
+
+    async fn find_enabled(&self) -> Result<Vec<Plugin>, DomainError> {
+        let rows = sqlx::query_as::<_, PluginRow>(
+            "SELECT plugin_id, name, plugin_type, is_enabled, configuration
+             FROM plugins WHERE is_enabled = true",
+        )
+        .fetch_all(&self.pool)
+        .await
+        .map_err(|e| DomainError::Internal(e.to_string()))?;
+
+        Ok(rows.into_iter().map(Into::into).collect())
+    }
+
+    async fn save(&self, plugin: &Plugin) -> Result<(), DomainError> {
+        sqlx::query(
+            "INSERT INTO plugins (plugin_id, name, plugin_type, is_enabled, configuration)
+             VALUES ($1, $2, $3, $4, $5)
+             ON CONFLICT (plugin_id) DO UPDATE SET
+               name = EXCLUDED.name,
+               plugin_type = EXCLUDED.plugin_type,
+               is_enabled = EXCLUDED.is_enabled,
+               configuration = EXCLUDED.configuration",
+        )
+        .bind(*plugin.plugin_id.as_uuid())
+        .bind(&plugin.name)
+        .bind(plugin_type_to_str(&plugin.plugin_type))
+        .bind(plugin.is_enabled)
+        .bind(structured_to_json(&plugin.configuration))
+        .execute(&self.pool)
+        .await
+        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        Ok(())
+    }
+}
+
+// ──────────────────────────────────────────────
+// Pipeline
+// ──────────────────────────────────────────────
+
+#[derive(sqlx::FromRow)]
+struct PipelineRow {
+    pipeline_id: Uuid,
+    trigger_event: String,
+    steps: serde_json::Value,
+}
+
+#[derive(serde::Serialize, serde::Deserialize)]
+struct StepJson {
+    plugin_id: Uuid,
+    step_order: u32,
+    configuration: serde_json::Map<String, serde_json::Value>,
+}
+
+fn steps_from_json(v: serde_json::Value) -> Vec<PipelineStep> {
+    let arr: Vec<StepJson> = serde_json::from_value(v).unwrap_or_default();
+    arr.into_iter()
+        .map(|s| {
+            let mut config = StructuredData::new();
+            for (k, val) in s.configuration {
+                config.insert(k, MetadataValue::from(val));
+            }
+            PipelineStep {
+                plugin_id: SystemId::from_uuid(s.plugin_id),
+                step_order: s.step_order,
+                configuration: config,
+            }
+        })
+        .collect()
+}
+
+fn steps_to_json(steps: &[PipelineStep]) -> serde_json::Value {
+    let arr: Vec<StepJson> = steps
+        .iter()
+        .map(|s| {
+            let config: serde_json::Map<String, serde_json::Value> = s
+                .configuration
+                .inner()
+                .iter()
+                .map(|(k, v)| (k.clone(), serde_json::Value::from(v)))
+                .collect();
+            StepJson {
+                plugin_id: *s.plugin_id.as_uuid(),
+                step_order: s.step_order,
+                configuration: config,
+            }
+        })
+        .collect();
+    serde_json::to_value(arr).unwrap_or(serde_json::Value::Array(vec![]))
+}
+
+impl From<PipelineRow> for ProcessingPipeline {
+    fn from(r: PipelineRow) -> Self {
+        Self {
+            pipeline_id: SystemId::from_uuid(r.pipeline_id),
+            trigger_event: r.trigger_event,
+            steps: steps_from_json(r.steps),
+        }
+    }
+}
+
+pub struct PostgresPipelineRepository {
+    pool: PgPool,
+}
+
+impl PostgresPipelineRepository {
+    pub fn new(pool: PgPool) -> Self {
+        Self { pool }
+    }
+}
+
+#[async_trait]
+impl PipelineRepository for PostgresPipelineRepository {
+    async fn find_by_id(&self, id: &SystemId) -> Result<Option<ProcessingPipeline>, DomainError> {
+        let row = sqlx::query_as::<_, PipelineRow>(
+            "SELECT pipeline_id, trigger_event, steps
+             FROM processing_pipelines WHERE pipeline_id = $1",
+        )
+        .bind(*id.as_uuid())
+        .fetch_optional(&self.pool)
+        .await
+        .map_err(|e| DomainError::Internal(e.to_string()))?;
+
+        Ok(row.map(Into::into))
+    }
+
+    async fn find_by_trigger(&self, event: &str) -> Result<Vec<ProcessingPipeline>, DomainError> {
+        let rows = sqlx::query_as::<_, PipelineRow>(
+            "SELECT pipeline_id, trigger_event, steps
+             FROM processing_pipelines WHERE trigger_event = $1",
+        )
+        .bind(event)
+        .fetch_all(&self.pool)
+        .await
+        .map_err(|e| DomainError::Internal(e.to_string()))?;
+
+        Ok(rows.into_iter().map(Into::into).collect())
+    }
+
+    async fn save(&self, pipeline: &ProcessingPipeline) -> Result<(), DomainError> {
+        sqlx::query(
+            "INSERT INTO processing_pipelines (pipeline_id, trigger_event, steps)
+             VALUES ($1, $2, $3)
+             ON CONFLICT (pipeline_id) DO UPDATE SET
+               trigger_event = EXCLUDED.trigger_event,
+               steps = EXCLUDED.steps",
+        )
+        .bind(*pipeline.pipeline_id.as_uuid())
+        .bind(&pipeline.trigger_event)
+        .bind(steps_to_json(&pipeline.steps))
+        .execute(&self.pool)
+        .await
+        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        Ok(())
+    }
+}