cooklang_sync_client/
lib.rs

1use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine};
2use futures::{channel::mpsc::channel, try_join};
3use notify::RecursiveMode;
4use serde::Deserialize;
5use std::path::PathBuf;
6use std::sync::Arc;
7use tokio::runtime::Runtime;
8use tokio::sync::Mutex;
9
10use log::debug;
11
12use crate::chunker::{Chunker, InMemoryCache};
13use crate::file_watcher::async_watcher;
14use crate::indexer::check_index_once;
15use crate::syncer::{check_download_once, check_upload_once};
16
17const CHANNEL_SIZE: usize = 100;
18const INMEMORY_CACHE_MAX_REC: usize = 100000;
19const INMEMORY_CACHE_MAX_MEM: u64 = 100_000_000_000;
20
21pub mod chunker;
22pub mod connection;
23pub mod context;
24pub mod errors;
25pub mod file_watcher;
26pub mod indexer;
27pub mod models;
28pub mod registry;
29pub mod remote;
30pub mod schema;
31pub mod syncer;
32
33// Export SyncStatus and context types for external use
34pub use context::{SyncContext, SyncStatusListener};
35pub use models::SyncStatus;
36
37/// Extracts the user ID from a JWT token without signature verification.
38/// JWT format: header.payload.signature (base64url encoded)
39pub fn extract_uid_from_jwt(token: &str) -> i32 {
40    #[derive(Deserialize)]
41    struct Claims {
42        uid: i32,
43    }
44
45    let parts: Vec<&str> = token.split('.').collect();
46    let payload = URL_SAFE_NO_PAD
47        .decode(parts[1])
48        .expect("Failed to decode JWT payload");
49    let claims: Claims = serde_json::from_slice(&payload).expect("Failed to parse JWT claims");
50
51    claims.uid
52}
53
54#[cfg(feature = "ffi")]
55uniffi::setup_scaffolding!();
56
57/// Synchronous alias to async run function.
58/// Intended to used by external (written in other languages) callers.
59#[cfg_attr(feature = "ffi", uniffi::export)]
60pub fn run(
61    context: Arc<SyncContext>,
62    storage_dir: &str,
63    db_file_path: &str,
64    api_endpoint: &str,
65    remote_token: &str,
66    namespace_id: i32,
67    download_only: bool,
68) -> Result<(), errors::SyncError> {
69    Runtime::new()?.block_on(run_async(
70        context,
71        storage_dir,
72        db_file_path,
73        api_endpoint,
74        remote_token,
75        namespace_id,
76        download_only,
77    ))?;
78
79    Ok(())
80}
81
82/// Connects to the server and waits either when `wait_time` expires or
83/// when there's a remote update for this client.
84/// Note, it doesn't do the update itself, you need to use `run_download_once`
85/// after this function completes.
86#[cfg_attr(feature = "ffi", uniffi::export)]
87pub fn wait_remote_update(api_endpoint: &str, remote_token: &str) -> Result<(), errors::SyncError> {
88    Runtime::new()?.block_on(remote::Remote::new(api_endpoint, remote_token).poll())?;
89
90    Ok(())
91}
92
93/// Runs one-off download of updates from remote server.
94/// Note, it's not very efficient as requires to re-initialize DB connection,
95/// chunker, remote client, etc every time it runs.
96#[cfg_attr(feature = "ffi", uniffi::export)]
97pub fn run_download_once(
98    storage_dir: &str,
99    db_file_path: &str,
100    api_endpoint: &str,
101    remote_token: &str,
102    namespace_id: i32,
103) -> Result<(), errors::SyncError> {
104    use std::env;
105
106    env::set_var("CARGO_LOG", "trace");
107
108    let storage_dir = &PathBuf::from(storage_dir);
109    let chunk_cache = InMemoryCache::new(INMEMORY_CACHE_MAX_REC, INMEMORY_CACHE_MAX_MEM);
110    let chunker = Arc::new(Mutex::new(Chunker::new(chunk_cache, storage_dir.clone())));
111    let remote = &remote::Remote::new(api_endpoint, remote_token);
112
113    let pool = connection::get_connection_pool(db_file_path)?;
114    debug!("Started connection pool for {:?}", db_file_path);
115
116    Runtime::new()?.block_on(check_download_once(
117        &pool,
118        Arc::clone(&chunker),
119        remote,
120        storage_dir,
121        namespace_id,
122    ))?;
123
124    Ok(())
125}
126
127/// Runs one-off upload of updates to remote server.
128/// Note, it's not very efficient as requires to re-initialize DB connection,
129/// chunker, remote client, etc every time it runs.
130#[cfg_attr(feature = "ffi", uniffi::export)]
131pub fn run_upload_once(
132    storage_dir: &str,
133    db_file_path: &str,
134    api_endpoint: &str,
135    remote_token: &str,
136    namespace_id: i32,
137) -> Result<(), errors::SyncError> {
138    let storage_dir = &PathBuf::from(storage_dir);
139    let chunk_cache = InMemoryCache::new(INMEMORY_CACHE_MAX_REC, INMEMORY_CACHE_MAX_MEM);
140    let chunker = Arc::new(Mutex::new(Chunker::new(chunk_cache, storage_dir.clone())));
141    let remote = &remote::Remote::new(api_endpoint, remote_token);
142
143    let pool = connection::get_connection_pool(db_file_path)?;
144    debug!("Started connection pool for {:?}", db_file_path);
145
146    check_index_once(&pool, storage_dir, namespace_id)?;
147
148    let runtime = Runtime::new()?;
149
150    // It requires first pass to upload missing chunks and second to
151    // commit and update `jid` to local records.
152    if !runtime.block_on(check_upload_once(
153        &pool,
154        Arc::clone(&chunker),
155        remote,
156        namespace_id,
157    ))? {
158        runtime.block_on(check_upload_once(
159            &pool,
160            Arc::clone(&chunker),
161            remote,
162            namespace_id,
163        ))?;
164    }
165
166    Ok(())
167}
168
169/// Runs local files watch and sync from/to remote continuously.
170#[allow(clippy::too_many_arguments)]
171pub async fn run_async(
172    context: Arc<SyncContext>,
173    storage_dir: &str,
174    db_file_path: &str,
175    api_endpoint: &str,
176    remote_token: &str,
177    namespace_id: i32,
178    download_only: bool,
179) -> Result<(), errors::SyncError> {
180    let token = context.token();
181    let listener = context.listener();
182
183    // Initialize all components first
184    let (mut debouncer, local_file_update_rx) = async_watcher()?;
185    let (local_registry_updated_tx, local_registry_updated_rx) = channel(CHANNEL_SIZE);
186
187    let storage_dir = &PathBuf::from(storage_dir);
188    let chunk_cache = InMemoryCache::new(INMEMORY_CACHE_MAX_REC, INMEMORY_CACHE_MAX_MEM);
189    let chunker = Chunker::new(chunk_cache, storage_dir.clone());
190    let remote = &remote::Remote::new(api_endpoint, remote_token);
191
192    let pool = connection::get_connection_pool(db_file_path)?;
193    debug!("Started connection pool for {:?}", db_file_path);
194
195    if !download_only {
196        debouncer
197            .watcher()
198            .watch(storage_dir, RecursiveMode::Recursive)?;
199        debug!("Started watcher on {:?}", storage_dir);
200    }
201
202    // Notify syncing status after successful initialization
203    if let Some(ref cb) = listener {
204        cb.on_status_changed(SyncStatus::Syncing);
205    }
206
207    let indexer = indexer::run(
208        token.clone(),
209        listener.clone(),
210        &pool,
211        storage_dir,
212        namespace_id,
213        local_file_update_rx,
214        local_registry_updated_tx,
215    );
216    debug!("Started indexer on {:?}", storage_dir);
217
218    let syncer = syncer::run(
219        token.clone(),
220        listener.clone(),
221        &pool,
222        storage_dir,
223        namespace_id,
224        chunker,
225        remote,
226        local_registry_updated_rx,
227        download_only,
228    );
229    debug!("Started syncer");
230
231    let result = try_join!(indexer, syncer);
232
233    // Notify completion (on_complete includes success status and optional error message)
234    if let Some(ref cb) = listener {
235        match result {
236            Ok(_) => cb.on_complete(true, None),
237            Err(ref e) => cb.on_complete(false, Some(format!("{:?}", e))),
238        }
239    }
240
241    result?;
242    Ok(())
243}
cooklang_sync_client/lib.rs

cooklang_sync_client/
lib.rs
