helios_persistence/core/
bulk_export.rs

1//! Bulk export types and traits.
2//!
3//! This module provides types and traits for implementing FHIR Bulk Data Export
4//! as specified in the [FHIR Bulk Data Access IG](https://hl7.org/fhir/uv/bulkdata/export.html).
5//!
6//! # Export Levels
7//!
8//! The Bulk Data Export specification supports three levels of export:
9//!
10//! - **System-level** (`[base]/$export`) - Exports all resources in the system
11//! - **Patient-level** (`[base]/Patient/$export`) - Exports all patient compartment resources
12//! - **Group-level** (`[base]/Group/[id]/$export`) - Exports resources for patients in a group
13//!
14//! # Example
15//!
16//! ```ignore
17//! use helios_persistence::core::bulk_export::{
18//!     BulkExportStorage, ExportRequest, ExportLevel, ExportStatus,
19//! };
20//!
21//! async fn export_patients<S: BulkExportStorage>(storage: &S, tenant: &TenantContext) {
22//!     // Start a system-level export of Patient resources
23//!     let request = ExportRequest::new(ExportLevel::System)
24//!         .with_types(vec!["Patient".to_string()]);
25//!
26//!     let job_id = storage.start_export(tenant, request).await.unwrap();
27//!
28//!     // Poll for completion
29//!     loop {
30//!         let progress = storage.get_export_status(tenant, &job_id).await.unwrap();
31//!         match progress.status {
32//!             ExportStatus::Complete => break,
33//!             ExportStatus::Error => panic!("Export failed"),
34//!             _ => tokio::time::sleep(std::time::Duration::from_secs(1)).await,
35//!         }
36//!     }
37//!
38//!     // Get the manifest
39//!     let manifest = storage.get_export_manifest(tenant, &job_id).await.unwrap();
40//!     for file in manifest.output {
41//!         println!("Exported {} {} resources to {}", file.count, file.resource_type, file.url);
42//!     }
43//! }
44//! ```
45
46use async_trait::async_trait;
47use chrono::{DateTime, Utc};
48use serde::{Deserialize, Serialize};
49use serde_json::Value;
50use uuid::Uuid;
51
52use crate::error::StorageResult;
53use crate::tenant::TenantContext;
54
55/// Unique identifier for an export job.
56#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
57pub struct ExportJobId(String);
58
59impl ExportJobId {
60    /// Creates a new random export job ID.
61    pub fn new() -> Self {
62        Self(Uuid::new_v4().to_string())
63    }
64
65    /// Creates an export job ID from an existing string.
66    pub fn from_string(id: impl Into<String>) -> Self {
67        Self(id.into())
68    }
69
70    /// Returns the ID as a string reference.
71    pub fn as_str(&self) -> &str {
72        &self.0
73    }
74}
75
76impl Default for ExportJobId {
77    fn default() -> Self {
78        Self::new()
79    }
80}
81
82impl std::fmt::Display for ExportJobId {
83    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
84        write!(f, "{}", self.0)
85    }
86}
87
88impl From<String> for ExportJobId {
89    fn from(s: String) -> Self {
90        Self(s)
91    }
92}
93
94impl From<&str> for ExportJobId {
95    fn from(s: &str) -> Self {
96        Self(s.to_string())
97    }
98}
99
100/// Status of an export job.
101#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
102#[serde(rename_all = "lowercase")]
103pub enum ExportStatus {
104    /// Job has been accepted but not yet started processing.
105    Accepted,
106    /// Job is currently processing.
107    InProgress,
108    /// Job has completed successfully.
109    Complete,
110    /// Job failed with an error.
111    Error,
112    /// Job was cancelled by the user.
113    Cancelled,
114}
115
116impl ExportStatus {
117    /// Returns true if the job is in a terminal state (complete, error, or cancelled).
118    pub fn is_terminal(&self) -> bool {
119        matches!(self, Self::Complete | Self::Error | Self::Cancelled)
120    }
121
122    /// Returns true if the job is still active (accepted or in progress).
123    pub fn is_active(&self) -> bool {
124        matches!(self, Self::Accepted | Self::InProgress)
125    }
126}
127
128impl std::fmt::Display for ExportStatus {
129    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
130        match self {
131            Self::Accepted => write!(f, "accepted"),
132            Self::InProgress => write!(f, "in-progress"),
133            Self::Complete => write!(f, "complete"),
134            Self::Error => write!(f, "error"),
135            Self::Cancelled => write!(f, "cancelled"),
136        }
137    }
138}
139
140impl std::str::FromStr for ExportStatus {
141    type Err = String;
142
143    fn from_str(s: &str) -> Result<Self, Self::Err> {
144        match s {
145            "accepted" => Ok(Self::Accepted),
146            "in-progress" | "in_progress" => Ok(Self::InProgress),
147            "complete" => Ok(Self::Complete),
148            "error" => Ok(Self::Error),
149            "cancelled" => Ok(Self::Cancelled),
150            _ => Err(format!("unknown export status: {}", s)),
151        }
152    }
153}
154
155/// Level at which the export is being performed.
156#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
157#[serde(rename_all = "lowercase")]
158pub enum ExportLevel {
159    /// System-level export (`[base]/$export`).
160    System,
161    /// Patient-level export (`[base]/Patient/$export`).
162    Patient,
163    /// Group-level export (`[base]/Group/[id]/$export`).
164    Group {
165        /// The group ID to export.
166        group_id: String,
167    },
168}
169
170impl ExportLevel {
171    /// Creates a system-level export.
172    pub fn system() -> Self {
173        Self::System
174    }
175
176    /// Creates a patient-level export.
177    pub fn patient() -> Self {
178        Self::Patient
179    }
180
181    /// Creates a group-level export for the given group ID.
182    pub fn group(group_id: impl Into<String>) -> Self {
183        Self::Group {
184            group_id: group_id.into(),
185        }
186    }
187}
188
189impl std::fmt::Display for ExportLevel {
190    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
191        match self {
192            Self::System => write!(f, "system"),
193            Self::Patient => write!(f, "patient"),
194            Self::Group { group_id } => write!(f, "group/{}", group_id),
195        }
196    }
197}
198
199/// A type filter for the export request.
200///
201/// Type filters allow specifying FHIR search parameters that should be applied
202/// when exporting a specific resource type.
203#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
204pub struct TypeFilter {
205    /// The resource type this filter applies to.
206    pub resource_type: String,
207    /// The search query parameters.
208    pub query: String,
209}
210
211impl TypeFilter {
212    /// Creates a new type filter.
213    pub fn new(resource_type: impl Into<String>, query: impl Into<String>) -> Self {
214        Self {
215            resource_type: resource_type.into(),
216            query: query.into(),
217        }
218    }
219}
220
221/// Request parameters for starting an export job.
222#[derive(Debug, Clone, Serialize, Deserialize)]
223pub struct ExportRequest {
224    /// The level at which to perform the export.
225    pub level: ExportLevel,
226
227    /// Resource types to export. If empty, all applicable types are exported.
228    #[serde(default)]
229    pub resource_types: Vec<String>,
230
231    /// Only include resources modified since this time.
232    #[serde(skip_serializing_if = "Option::is_none")]
233    pub since: Option<DateTime<Utc>>,
234
235    /// Type-specific filters to apply during export.
236    #[serde(default)]
237    pub type_filters: Vec<TypeFilter>,
238
239    /// Batch size for processing (implementation-specific).
240    #[serde(default = "default_batch_size")]
241    pub batch_size: u32,
242
243    /// Output format (default: "application/fhir+ndjson").
244    #[serde(default = "default_output_format")]
245    pub output_format: String,
246}
247
248fn default_batch_size() -> u32 {
249    1000
250}
251
252fn default_output_format() -> String {
253    "application/fhir+ndjson".to_string()
254}
255
256impl ExportRequest {
257    /// Creates a new export request with the given level.
258    pub fn new(level: ExportLevel) -> Self {
259        Self {
260            level,
261            resource_types: Vec::new(),
262            since: None,
263            type_filters: Vec::new(),
264            batch_size: default_batch_size(),
265            output_format: default_output_format(),
266        }
267    }
268
269    /// Creates a system-level export request.
270    pub fn system() -> Self {
271        Self::new(ExportLevel::System)
272    }
273
274    /// Creates a patient-level export request.
275    pub fn patient() -> Self {
276        Self::new(ExportLevel::Patient)
277    }
278
279    /// Creates a group-level export request.
280    pub fn group(group_id: impl Into<String>) -> Self {
281        Self::new(ExportLevel::Group {
282            group_id: group_id.into(),
283        })
284    }
285
286    /// Sets the resource types to export.
287    pub fn with_types(mut self, types: Vec<String>) -> Self {
288        self.resource_types = types;
289        self
290    }
291
292    /// Sets the since filter.
293    pub fn with_since(mut self, since: DateTime<Utc>) -> Self {
294        self.since = Some(since);
295        self
296    }
297
298    /// Adds a type filter.
299    pub fn with_type_filter(mut self, filter: TypeFilter) -> Self {
300        self.type_filters.push(filter);
301        self
302    }
303
304    /// Adds multiple type filters.
305    pub fn with_type_filters(mut self, filters: Vec<TypeFilter>) -> Self {
306        self.type_filters.extend(filters);
307        self
308    }
309
310    /// Sets the batch size.
311    pub fn with_batch_size(mut self, batch_size: u32) -> Self {
312        self.batch_size = batch_size;
313        self
314    }
315
316    /// Returns the group ID if this is a group-level export.
317    pub fn group_id(&self) -> Option<&str> {
318        match &self.level {
319            ExportLevel::Group { group_id } => Some(group_id),
320            _ => None,
321        }
322    }
323}
324
325/// Progress information for a single resource type in an export.
326#[derive(Debug, Clone, Serialize, Deserialize)]
327pub struct TypeExportProgress {
328    /// The resource type.
329    pub resource_type: String,
330    /// Total number of resources to export (may be estimated).
331    pub total_count: Option<u64>,
332    /// Number of resources exported so far.
333    pub exported_count: u64,
334    /// Number of errors encountered.
335    pub error_count: u64,
336    /// Current cursor state for resuming (opaque to clients).
337    #[serde(skip_serializing_if = "Option::is_none")]
338    pub cursor_state: Option<String>,
339}
340
341impl TypeExportProgress {
342    /// Creates new progress tracking for a resource type.
343    pub fn new(resource_type: impl Into<String>) -> Self {
344        Self {
345            resource_type: resource_type.into(),
346            total_count: None,
347            exported_count: 0,
348            error_count: 0,
349            cursor_state: None,
350        }
351    }
352
353    /// Sets the total count.
354    pub fn with_total(mut self, total: u64) -> Self {
355        self.total_count = Some(total);
356        self
357    }
358
359    /// Returns the progress as a percentage (0.0 to 1.0).
360    pub fn progress_fraction(&self) -> Option<f64> {
361        self.total_count.map(|total| {
362            if total == 0 {
363                1.0
364            } else {
365                self.exported_count as f64 / total as f64
366            }
367        })
368    }
369}
370
371/// Overall progress of an export job.
372#[derive(Debug, Clone, Serialize, Deserialize)]
373pub struct ExportProgress {
374    /// The job ID.
375    pub job_id: ExportJobId,
376    /// Current status of the job.
377    pub status: ExportStatus,
378    /// The export level.
379    pub level: ExportLevel,
380    /// Time the export was initiated.
381    pub transaction_time: DateTime<Utc>,
382    /// Time the export started processing.
383    #[serde(skip_serializing_if = "Option::is_none")]
384    pub started_at: Option<DateTime<Utc>>,
385    /// Time the export completed (success, error, or cancelled).
386    #[serde(skip_serializing_if = "Option::is_none")]
387    pub completed_at: Option<DateTime<Utc>>,
388    /// Per-type progress information.
389    pub type_progress: Vec<TypeExportProgress>,
390    /// Current type being processed.
391    #[serde(skip_serializing_if = "Option::is_none")]
392    pub current_type: Option<String>,
393    /// Error message if status is Error.
394    #[serde(skip_serializing_if = "Option::is_none")]
395    pub error_message: Option<String>,
396}
397
398impl ExportProgress {
399    /// Creates new progress for an accepted job.
400    pub fn accepted(
401        job_id: ExportJobId,
402        level: ExportLevel,
403        transaction_time: DateTime<Utc>,
404    ) -> Self {
405        Self {
406            job_id,
407            status: ExportStatus::Accepted,
408            level,
409            transaction_time,
410            started_at: None,
411            completed_at: None,
412            type_progress: Vec::new(),
413            current_type: None,
414            error_message: None,
415        }
416    }
417
418    /// Returns the overall progress as a percentage (0.0 to 1.0).
419    pub fn overall_progress(&self) -> f64 {
420        if self.type_progress.is_empty() {
421            return 0.0;
422        }
423
424        let (total_exported, total_count) = self
425            .type_progress
426            .iter()
427            .fold((0u64, 0u64), |(exp, tot), tp| {
428                (exp + tp.exported_count, tot + tp.total_count.unwrap_or(0))
429            });
430
431        if total_count == 0 {
432            0.0
433        } else {
434            total_exported as f64 / total_count as f64
435        }
436    }
437}
438
439/// An output file in the export manifest.
440#[derive(Debug, Clone, Serialize, Deserialize)]
441pub struct ExportOutputFile {
442    /// The resource type contained in this file.
443    #[serde(rename = "type")]
444    pub resource_type: String,
445    /// URL to access the file.
446    pub url: String,
447    /// Number of resources in the file.
448    #[serde(skip_serializing_if = "Option::is_none")]
449    pub count: Option<u64>,
450}
451
452impl ExportOutputFile {
453    /// Creates a new output file descriptor.
454    pub fn new(resource_type: impl Into<String>, url: impl Into<String>) -> Self {
455        Self {
456            resource_type: resource_type.into(),
457            url: url.into(),
458            count: None,
459        }
460    }
461
462    /// Sets the count.
463    pub fn with_count(mut self, count: u64) -> Self {
464        self.count = Some(count);
465        self
466    }
467}
468
469/// The export manifest returned when an export completes.
470///
471/// This follows the FHIR Bulk Data Export manifest format.
472#[derive(Debug, Clone, Serialize, Deserialize)]
473pub struct ExportManifest {
474    /// Time the export was initiated.
475    #[serde(rename = "transactionTime")]
476    pub transaction_time: DateTime<Utc>,
477    /// The original export request URL.
478    pub request: String,
479    /// Whether the client should check for deleted resources.
480    #[serde(rename = "requiresAccessToken")]
481    pub requires_access_token: bool,
482    /// Output files containing the exported resources.
483    pub output: Vec<ExportOutputFile>,
484    /// Output files containing OperationOutcome resources for errors.
485    #[serde(default)]
486    pub error: Vec<ExportOutputFile>,
487    /// Informational messages.
488    #[serde(default, skip_serializing_if = "Option::is_none")]
489    pub message: Option<String>,
490    /// Extension data.
491    #[serde(default, skip_serializing_if = "Option::is_none")]
492    pub extension: Option<Value>,
493}
494
495impl ExportManifest {
496    /// Creates a new export manifest.
497    pub fn new(transaction_time: DateTime<Utc>, request: impl Into<String>) -> Self {
498        Self {
499            transaction_time,
500            request: request.into(),
501            requires_access_token: true,
502            output: Vec::new(),
503            error: Vec::new(),
504            message: None,
505            extension: None,
506        }
507    }
508
509    /// Adds an output file.
510    pub fn with_output(mut self, file: ExportOutputFile) -> Self {
511        self.output.push(file);
512        self
513    }
514
515    /// Adds an error file.
516    pub fn with_error(mut self, file: ExportOutputFile) -> Self {
517        self.error.push(file);
518        self
519    }
520
521    /// Sets a message.
522    pub fn with_message(mut self, message: impl Into<String>) -> Self {
523        self.message = Some(message.into());
524        self
525    }
526}
527
528/// A batch of NDJSON resources for streaming export.
529#[derive(Debug, Clone)]
530pub struct NdjsonBatch {
531    /// The serialized NDJSON lines (one JSON object per line).
532    pub lines: Vec<String>,
533    /// Cursor for fetching the next batch, if any.
534    pub next_cursor: Option<String>,
535    /// Whether this is the last batch.
536    pub is_last: bool,
537}
538
539impl NdjsonBatch {
540    /// Creates a new batch.
541    pub fn new(lines: Vec<String>) -> Self {
542        Self {
543            lines,
544            next_cursor: None,
545            is_last: false,
546        }
547    }
548
549    /// Creates an empty final batch.
550    pub fn empty() -> Self {
551        Self {
552            lines: Vec::new(),
553            next_cursor: None,
554            is_last: true,
555        }
556    }
557
558    /// Sets the next cursor.
559    pub fn with_cursor(mut self, cursor: impl Into<String>) -> Self {
560        self.next_cursor = Some(cursor.into());
561        self
562    }
563
564    /// Marks this as the last batch.
565    pub fn as_last(mut self) -> Self {
566        self.is_last = true;
567        self.next_cursor = None;
568        self
569    }
570
571    /// Returns the number of resources in this batch.
572    pub fn len(&self) -> usize {
573        self.lines.len()
574    }
575
576    /// Returns true if this batch is empty.
577    pub fn is_empty(&self) -> bool {
578        self.lines.is_empty()
579    }
580}
581
582// ============================================================================
583// Traits
584// ============================================================================
585
586/// Storage trait for bulk export job management.
587///
588/// This trait handles the lifecycle of export jobs: creating, tracking,
589/// completing, and cleaning up exports.
590#[async_trait]
591pub trait BulkExportStorage: Send + Sync {
592    /// Starts a new export job.
593    ///
594    /// # Arguments
595    ///
596    /// * `tenant` - The tenant context
597    /// * `request` - The export request parameters
598    ///
599    /// # Returns
600    ///
601    /// The job ID for tracking the export.
602    ///
603    /// # Errors
604    ///
605    /// * `BulkExportError::TooManyConcurrentExports` - If too many exports are running
606    /// * `BulkExportError::InvalidRequest` - If the request is invalid
607    async fn start_export(
608        &self,
609        tenant: &TenantContext,
610        request: ExportRequest,
611    ) -> StorageResult<ExportJobId>;
612
613    /// Gets the current status of an export job.
614    ///
615    /// # Arguments
616    ///
617    /// * `tenant` - The tenant context
618    /// * `job_id` - The export job ID
619    ///
620    /// # Returns
621    ///
622    /// The current progress of the export.
623    ///
624    /// # Errors
625    ///
626    /// * `BulkExportError::JobNotFound` - If the job doesn't exist
627    async fn get_export_status(
628        &self,
629        tenant: &TenantContext,
630        job_id: &ExportJobId,
631    ) -> StorageResult<ExportProgress>;
632
633    /// Cancels an in-progress export job.
634    ///
635    /// # Arguments
636    ///
637    /// * `tenant` - The tenant context
638    /// * `job_id` - The export job ID
639    ///
640    /// # Errors
641    ///
642    /// * `BulkExportError::JobNotFound` - If the job doesn't exist
643    /// * `BulkExportError::InvalidJobState` - If the job is already complete
644    async fn cancel_export(
645        &self,
646        tenant: &TenantContext,
647        job_id: &ExportJobId,
648    ) -> StorageResult<()>;
649
650    /// Deletes an export job and its output files.
651    ///
652    /// # Arguments
653    ///
654    /// * `tenant` - The tenant context
655    /// * `job_id` - The export job ID
656    ///
657    /// # Errors
658    ///
659    /// * `BulkExportError::JobNotFound` - If the job doesn't exist
660    async fn delete_export(
661        &self,
662        tenant: &TenantContext,
663        job_id: &ExportJobId,
664    ) -> StorageResult<()>;
665
666    /// Gets the manifest for a completed export.
667    ///
668    /// # Arguments
669    ///
670    /// * `tenant` - The tenant context
671    /// * `job_id` - The export job ID
672    ///
673    /// # Returns
674    ///
675    /// The export manifest with output file information.
676    ///
677    /// # Errors
678    ///
679    /// * `BulkExportError::JobNotFound` - If the job doesn't exist
680    /// * `BulkExportError::InvalidJobState` - If the job is not complete
681    async fn get_export_manifest(
682        &self,
683        tenant: &TenantContext,
684        job_id: &ExportJobId,
685    ) -> StorageResult<ExportManifest>;
686
687    /// Lists export jobs for a tenant.
688    ///
689    /// # Arguments
690    ///
691    /// * `tenant` - The tenant context
692    /// * `include_completed` - Whether to include completed jobs
693    ///
694    /// # Returns
695    ///
696    /// List of export progress records.
697    async fn list_exports(
698        &self,
699        tenant: &TenantContext,
700        include_completed: bool,
701    ) -> StorageResult<Vec<ExportProgress>>;
702}
703
704/// Data provider for export operations.
705///
706/// This trait provides the data retrieval capabilities needed to perform
707/// system-level exports.
708#[async_trait]
709pub trait ExportDataProvider: Send + Sync {
710    /// Lists resource types available for export.
711    ///
712    /// # Arguments
713    ///
714    /// * `tenant` - The tenant context
715    /// * `request` - The export request (used to filter by requested types)
716    ///
717    /// # Returns
718    ///
719    /// List of resource type names that should be exported.
720    async fn list_export_types(
721        &self,
722        tenant: &TenantContext,
723        request: &ExportRequest,
724    ) -> StorageResult<Vec<String>>;
725
726    /// Counts resources of a type for export.
727    ///
728    /// # Arguments
729    ///
730    /// * `tenant` - The tenant context
731    /// * `request` - The export request (for filters)
732    /// * `resource_type` - The resource type to count
733    ///
734    /// # Returns
735    ///
736    /// The count of resources matching the export criteria.
737    async fn count_export_resources(
738        &self,
739        tenant: &TenantContext,
740        request: &ExportRequest,
741        resource_type: &str,
742    ) -> StorageResult<u64>;
743
744    /// Fetches a batch of resources for export.
745    ///
746    /// # Arguments
747    ///
748    /// * `tenant` - The tenant context
749    /// * `request` - The export request (for filters)
750    /// * `resource_type` - The resource type to fetch
751    /// * `cursor` - Cursor from previous batch, or None for first batch
752    /// * `batch_size` - Maximum number of resources to return
753    ///
754    /// # Returns
755    ///
756    /// A batch of NDJSON lines with cursor for next batch.
757    async fn fetch_export_batch(
758        &self,
759        tenant: &TenantContext,
760        request: &ExportRequest,
761        resource_type: &str,
762        cursor: Option<&str>,
763        batch_size: u32,
764    ) -> StorageResult<NdjsonBatch>;
765}
766
767/// Provider for patient compartment exports.
768///
769/// This trait extends `ExportDataProvider` with patient-specific capabilities
770/// needed for Patient-level exports.
771#[async_trait]
772pub trait PatientExportProvider: ExportDataProvider {
773    /// Lists patient IDs to include in the export.
774    ///
775    /// # Arguments
776    ///
777    /// * `tenant` - The tenant context
778    /// * `request` - The export request
779    /// * `cursor` - Cursor from previous call, or None for first call
780    /// * `batch_size` - Maximum number of patient IDs to return
781    ///
782    /// # Returns
783    ///
784    /// A tuple of (patient_ids, next_cursor).
785    async fn list_patient_ids(
786        &self,
787        tenant: &TenantContext,
788        request: &ExportRequest,
789        cursor: Option<&str>,
790        batch_size: u32,
791    ) -> StorageResult<(Vec<String>, Option<String>)>;
792
793    /// Fetches a batch of resources from the patient compartment.
794    ///
795    /// # Arguments
796    ///
797    /// * `tenant` - The tenant context
798    /// * `request` - The export request
799    /// * `resource_type` - The resource type to fetch
800    /// * `patient_ids` - Patient IDs whose resources to fetch
801    /// * `cursor` - Cursor from previous batch, or None for first batch
802    /// * `batch_size` - Maximum number of resources to return
803    ///
804    /// # Returns
805    ///
806    /// A batch of NDJSON lines with cursor for next batch.
807    async fn fetch_patient_compartment_batch(
808        &self,
809        tenant: &TenantContext,
810        request: &ExportRequest,
811        resource_type: &str,
812        patient_ids: &[String],
813        cursor: Option<&str>,
814        batch_size: u32,
815    ) -> StorageResult<NdjsonBatch>;
816}
817
818/// Provider for group-level exports.
819///
820/// This trait extends `PatientExportProvider` with group-specific capabilities
821/// needed for Group-level exports.
822#[async_trait]
823pub trait GroupExportProvider: PatientExportProvider {
824    /// Gets the members of a group.
825    ///
826    /// # Arguments
827    ///
828    /// * `tenant` - The tenant context
829    /// * `group_id` - The group resource ID
830    ///
831    /// # Returns
832    ///
833    /// List of member references (e.g., "Patient/123").
834    ///
835    /// # Errors
836    ///
837    /// * `BulkExportError::GroupNotFound` - If the group doesn't exist
838    async fn get_group_members(
839        &self,
840        tenant: &TenantContext,
841        group_id: &str,
842    ) -> StorageResult<Vec<String>>;
843
844    /// Resolves group members to patient IDs.
845    ///
846    /// This handles the case where group members may be references to other
847    /// resources (like Practitioner) or nested Groups.
848    ///
849    /// # Arguments
850    ///
851    /// * `tenant` - The tenant context
852    /// * `group_id` - The group resource ID
853    ///
854    /// # Returns
855    ///
856    /// List of patient IDs for the group members.
857    async fn resolve_group_patient_ids(
858        &self,
859        tenant: &TenantContext,
860        group_id: &str,
861    ) -> StorageResult<Vec<String>>;
862}
863
864#[cfg(test)]
865mod tests {
866    use super::*;
867
868    #[test]
869    fn test_export_job_id() {
870        let id = ExportJobId::new();
871        assert!(!id.as_str().is_empty());
872
873        let id2 = ExportJobId::from_string("test-123");
874        assert_eq!(id2.as_str(), "test-123");
875        assert_eq!(id2.to_string(), "test-123");
876    }
877
878    #[test]
879    fn test_export_status() {
880        assert!(ExportStatus::Complete.is_terminal());
881        assert!(ExportStatus::Error.is_terminal());
882        assert!(ExportStatus::Cancelled.is_terminal());
883        assert!(!ExportStatus::Accepted.is_terminal());
884        assert!(!ExportStatus::InProgress.is_terminal());
885
886        assert!(ExportStatus::Accepted.is_active());
887        assert!(ExportStatus::InProgress.is_active());
888        assert!(!ExportStatus::Complete.is_active());
889    }
890
891    #[test]
892    fn test_export_status_display_parse() {
893        let status = ExportStatus::InProgress;
894        assert_eq!(status.to_string(), "in-progress");
895
896        let parsed: ExportStatus = "in-progress".parse().unwrap();
897        assert_eq!(parsed, ExportStatus::InProgress);
898
899        // Also accept underscore variant
900        let parsed: ExportStatus = "in_progress".parse().unwrap();
901        assert_eq!(parsed, ExportStatus::InProgress);
902    }
903
904    #[test]
905    fn test_export_level() {
906        let system = ExportLevel::system();
907        assert!(matches!(system, ExportLevel::System));
908
909        let patient = ExportLevel::patient();
910        assert!(matches!(patient, ExportLevel::Patient));
911
912        let group = ExportLevel::group("grp-123");
913        assert!(matches!(group, ExportLevel::Group { group_id } if group_id == "grp-123"));
914    }
915
916    #[test]
917    fn test_export_request_builder() {
918        let request = ExportRequest::system()
919            .with_types(vec!["Patient".to_string(), "Observation".to_string()])
920            .with_batch_size(500)
921            .with_type_filter(TypeFilter::new("Observation", "code=1234"));
922
923        assert!(matches!(request.level, ExportLevel::System));
924        assert_eq!(request.resource_types, vec!["Patient", "Observation"]);
925        assert_eq!(request.batch_size, 500);
926        assert_eq!(request.type_filters.len(), 1);
927    }
928
929    #[test]
930    fn test_export_request_group_id() {
931        let request = ExportRequest::group("grp-123");
932        assert_eq!(request.group_id(), Some("grp-123"));
933
934        let system_request = ExportRequest::system();
935        assert_eq!(system_request.group_id(), None);
936    }
937
938    #[test]
939    fn test_type_export_progress() {
940        let progress = TypeExportProgress::new("Patient").with_total(100);
941        assert_eq!(progress.total_count, Some(100));
942        assert_eq!(progress.progress_fraction(), Some(0.0));
943
944        let mut progress = progress;
945        progress.exported_count = 50;
946        assert_eq!(progress.progress_fraction(), Some(0.5));
947    }
948
949    #[test]
950    fn test_export_manifest() {
951        let manifest = ExportManifest::new(Utc::now(), "https://example.com/$export")
952            .with_output(
953                ExportOutputFile::new("Patient", "/exports/Patient.ndjson").with_count(100),
954            )
955            .with_message("Export complete");
956
957        assert_eq!(manifest.output.len(), 1);
958        assert_eq!(manifest.output[0].resource_type, "Patient");
959        assert_eq!(manifest.output[0].count, Some(100));
960        assert!(manifest.message.is_some());
961    }
962
963    #[test]
964    fn test_ndjson_batch() {
965        let batch = NdjsonBatch::new(vec![
966            r#"{"resourceType":"Patient","id":"1"}"#.to_string(),
967            r#"{"resourceType":"Patient","id":"2"}"#.to_string(),
968        ])
969        .with_cursor("next-page");
970
971        assert_eq!(batch.len(), 2);
972        assert!(!batch.is_empty());
973        assert!(!batch.is_last);
974        assert_eq!(batch.next_cursor, Some("next-page".to_string()));
975
976        let final_batch = batch.as_last();
977        assert!(final_batch.is_last);
978        assert!(final_batch.next_cursor.is_none());
979    }
980
981    #[test]
982    fn test_ndjson_batch_empty() {
983        let batch = NdjsonBatch::empty();
984        assert!(batch.is_empty());
985        assert!(batch.is_last);
986    }
987}
helios_persistence/core/bulk_export.rs

helios_persistence/core/
bulk_export.rs
