surreal_sync_runtime/
lib.rs

1//! Shared runtime for surreal-sync: process setup, SurrealDB connection settings,
2//! transform loading, and the apply pipeline.
3//!
4//! Origin `from-*` crates re-export the pieces embedders need. The stock CLI
5//! may still auto-detect SurrealDB major version (linking both SDKs); the
6//! documented embed path uses a pre-built sink or `run::<OneSinkType>`.
7//!
8//! Sink crates use the default features (no CLI argument parsing). Enable `cli`
9//! only if you need [`SurrealCliOpts`].
10//!
11//! Enable `checkpoint_fs` (default) for [`checkpoint_fs::FilesystemStore`].
12//!
13//! # Sink + checkpoint typing
14//!
15//! - [`SinkConnect`] / [`SinkWithCheckpoints`] / [`SurrealConfig`] — defined in
16//!   [`surreal_sync_core`], re-exported here for existing imports
17//! - [`SinkWithCheckpoints`] opens a matching Surreal-table
18//!   [`CheckpointStore`](surreal_sync_core::CheckpointStore) for
19//!   `--checkpoints-surreal-table` (implemented on `Surreal2Sink` / `Surreal3Sink`)
20
21mod config;
22mod init;
23mod sink_connect;
24mod transforms;
25
26/// Shared apply loop, transform pipeline, and interleaved snapshot engine.
27pub mod pipeline;
28
29/// Store checkpoints on the filesystem.
30#[cfg(feature = "checkpoint_fs")]
31pub mod checkpoint_fs;
32
33#[cfg(feature = "cli")]
34mod cli_opts;
35
36#[cfg(feature = "cli")]
37pub use cli_opts::SurrealCliOpts;
38pub use config::SurrealConfig;
39pub use init::init;
40pub use sink_connect::{SinkConnect, SinkWithCheckpoints};
41pub use transforms::{load_transforms_from_args, merge_inplace_boxed, merge_inplace_transforms};
42
43// Re-exports commonly used pipeline types at the crate root.
44pub use pipeline::{
45    apply_changes, apply_changes_with, apply_relation_changes, apply_relation_changes_with,
46    ensure_command_resolvable, load_pipeline_and_opts, load_transforms_config, parse_humantime,
47    parse_transforms_toml, relation_wire_batch_id, run_adhoc_snapshot_tables,
48    run_adhoc_snapshot_tables_with_transforms, run_change_feed, run_change_feed_with,
49    run_interleaved_snapshot, run_interleaved_snapshot_with_resume,
50    run_interleaved_snapshot_with_resume_and_transforms, run_interleaved_snapshot_with_transforms,
51    run_source_runtime, run_source_runtime_with, write_relations, write_relations_with, write_rows,
52    write_rows_with, AdhocApply, ApplyContext, ApplyEvent, ApplyOpts, BatchTransformer, ChangeFeed,
53    ChangeFeedDriver, ChangeFeedRef, CheckpointPolicy, ChildStdioMode, CommandStageConfig,
54    ConfiguredStage, ControlSignal, CowBatch, ExternalTransform, ExternalTransport, FailurePolicy,
55    FlattenId, FlattenIdStageConfig, Framer, FramerKind, InPlaceTransform,
56    InterleavedSnapshotCheckpoint, InterleavedSnapshotConfig, InterleavedSnapshotResult,
57    ManagerCheckpointer, NdjsonFramer, NoopCheckpointer, Passthrough, PersistentChildStdio,
58    Pipeline, PipelineSection, PkTuple, PositionedChange, PositionedEvent, ReconciliationEvent,
59    ReconciliationPos, RelationChunkDriver, RelationChunkSource, RequestHeader, ResponseHeader,
60    RetryPolicy, RowChunkDriver, RowChunkSource, RuntimeExit, SnapshotCheckpointer, SnapshotSignal,
61    SnapshotTableProgress, SnapshotTransforms, SourceDriver, SourceRuntimeOpts, Stage, StdioConfig,
62    StopReason, TableSpec, TransformsConfig, TransientChildStdio, WatermarkKind, WatermarkSource,
63    WireItemKind, WireResponse, DEFAULT_CHUNK_SIZE, DEFAULT_FLATTEN_ID_SEPARATOR,
64    RELATION_WIRE_BATCH_ID_BIT,
65};
66
67#[cfg(any(test, feature = "test-support"))]
68pub use pipeline::test_support;
69
70/// Parse a duration string like "1h", "30m", "300s", "300" into seconds.
71pub fn parse_duration_to_secs(s: &str) -> anyhow::Result<i64> {
72    use anyhow::Context;
73
74    let s = s.trim();
75    if s.is_empty() {
76        anyhow::bail!("Empty duration string");
77    }
78
79    if let Some(num_str) = s.strip_suffix('h') {
80        let hours: i64 = num_str
81            .parse()
82            .with_context(|| format!("Invalid hours value: {num_str}"))?;
83        return Ok(hours * 3600);
84    }
85    if let Some(num_str) = s.strip_suffix('m') {
86        let minutes: i64 = num_str
87            .parse()
88            .with_context(|| format!("Invalid minutes value: {num_str}"))?;
89        return Ok(minutes * 60);
90    }
91    if let Some(num_str) = s.strip_suffix('s') {
92        let secs: i64 = num_str
93            .parse()
94            .with_context(|| format!("Invalid seconds value: {num_str}"))?;
95        return Ok(secs);
96    }
97
98    s.parse::<i64>()
99        .with_context(|| format!("Invalid duration value: {s}"))
100}
surreal_sync_runtime/lib.rs

surreal_sync_runtime/
lib.rs
