linkrs_server/batch/
types.rs

1//! Batch Operation Type Definition
2//!
3//! Wire DTOs are shared with the CLI through `linkrs-wire`; this module
4//! keeps only the server-internal [`BatchTask`] bookkeeping type.
5
6pub use linkrs_wire::batch::{
7    AddBatchItemsRequest, AddBatchItemsResponse, BatchErrorData, BatchId, BatchItem, BatchItemType,
8    BatchProgress, BatchResultData, BatchStatus, BatchStatusResponse, BatchType,
9    CreateBatchRequest, CreateBatchResponse, DeleteEdgeData, DeleteVertexData, EdgeData,
10    ExecuteBatchResponse, UpdateEdgeData, VertexData,
11};
12
13/// Batch task information (for internal use)
14#[derive(Debug, Clone)]
15pub struct BatchTask {
16    /// Task ID
17    pub id: BatchId,
18    /// Figure Space ID
19    pub space_id: u64,
20    /// Batch Type
21    pub batch_type: BatchType,
22    /// Batch size
23    pub batch_size: usize,
24    /// Task Status
25    pub status: BatchStatus,
26    /// Buffered items
27    pub buffered_items: Vec<BatchItem>,
28    /// Progress
29    pub progress: BatchProgress,
30    /// Result
31    pub result: Option<BatchResultData>,
32    /// Creation time
33    pub created_at: chrono::DateTime<chrono::Utc>,
34    /// Update time
35    pub updated_at: chrono::DateTime<chrono::Utc>,
36}
37
38impl BatchTask {
39    /// Create a new batch task.
40    pub fn new(id: BatchId, space_id: u64, batch_type: BatchType, batch_size: usize) -> Self {
41        let now = chrono::Utc::now();
42        Self {
43            id,
44            space_id,
45            batch_type,
46            batch_size,
47            status: BatchStatus::Created,
48            buffered_items: Vec::new(),
49            progress: BatchProgress {
50                total: 0,
51                processed: 0,
52                succeeded: 0,
53                failed: 0,
54                buffered: 0,
55            },
56            result: None,
57            created_at: now,
58            updated_at: now,
59        }
60    }
61
62    /// Update status
63    pub fn update_status(&mut self, status: BatchStatus) {
64        self.status = status;
65        self.updated_at = chrono::Utc::now();
66    }
67
68    /// Add a buffer item
69    pub fn add_items(&mut self, items: Vec<BatchItem>) -> usize {
70        let count = items.len();
71        self.buffered_items.extend(items);
72        self.progress.buffered = self.buffered_items.len();
73        self.progress.total += count;
74        self.updated_at = chrono::Utc::now();
75        count
76    }
77
78    /// Retrieve and clear the buffer items.
79    pub fn take_buffered_items(&mut self) -> Vec<BatchItem> {
80        let items = std::mem::take(&mut self.buffered_items);
81        self.progress.buffered = 0;
82        self.updated_at = chrono::Utc::now();
83        items
84    }
85
86    /// Update progress
87    pub fn update_progress(&mut self, succeeded: usize, failed: usize) {
88        self.progress.succeeded += succeeded;
89        self.progress.failed += failed;
90        self.progress.processed += succeeded + failed;
91        self.updated_at = chrono::Utc::now();
92    }
93
94    /// Set the results
95    pub fn set_result(&mut self, result: BatchResultData) {
96        self.result = Some(result);
97        self.updated_at = chrono::Utc::now();
98    }
99}
linkrs_server/batch/types.rs

linkrs_server/batch/
types.rs
