tegdb 0.5.0

The name TegridyDB (short for TegDB) is inspired by the Tegridy Farm in South Park and tries to correct some of the wrong database implementations, such as null support, implicit conversion support, etc.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138

use clap::{Parser, Subcommand};
use std::path::PathBuf;
use tegdb::tgstream::{Config, Restore, Result, S3Backend, Tailer};

#[derive(Parser)]
#[command(name = "tgstream")]
#[command(about = "Streaming backup tool for TegDB databases")]
#[command(version)]
struct Cli {
    #[command(subcommand)]
    command: Commands,
}

#[derive(Subcommand)]
enum Commands {
    /// Run continuous replication
    Run {
        /// Config file path
        #[arg(short, long)]
        config: PathBuf,
    },
    /// Create a one-off snapshot
    Snapshot {
        /// Config file path
        #[arg(short, long)]
        config: PathBuf,
    },
    /// Restore database from S3
    Restore {
        /// Config file path
        #[arg(short, long)]
        config: PathBuf,
        /// Output path for restored database
        #[arg(short, long)]
        to: PathBuf,
        /// Restore to specific offset (optional)
        #[arg(long)]
        at_offset: Option<u64>,
    },
    /// List available snapshots and segments
    List {
        /// Config file path
        #[arg(short, long)]
        config: PathBuf,
    },
    /// Prune old snapshots and segments
    Prune {
        /// Config file path
        #[arg(short, long)]
        config: PathBuf,
    },
}

#[tokio::main]
async fn main() {
    let cli = Cli::parse();

    if let Err(e) = run_command(cli.command).await {
        eprintln!("Error: {}", e);
        std::process::exit(1);
    }
}

async fn run_command(command: Commands) -> Result<()> {
    match command {
        Commands::Run { config } => {
            let config = Config::from_file(&config)?;
            let mut tailer = Tailer::new(config).await?;
            tailer.run().await?;
        }
        Commands::Snapshot { config } => {
            let config = Config::from_file(&config)?;
            let mut tailer = Tailer::new(config).await?;
            tailer.snapshot_once().await?;
        }
        Commands::Restore {
            config,
            to,
            at_offset,
        } => {
            let config = Config::from_file(&config)?;
            let restore = Restore::new(&config.s3, &config.s3_prefix()).await?;
            restore.restore_to(&to, at_offset).await?;
        }
        Commands::List { config } => {
            let config = Config::from_file(&config)?;
            let restore = Restore::new(&config.s3, &config.s3_prefix()).await?;
            restore.list().await?;
        }
        Commands::Prune { config } => {
            let config = Config::from_file(&config)?;
            prune_snapshots(&config).await?;
        }
    }

    Ok(())
}

async fn prune_snapshots(config: &Config) -> Result<()> {
    let backend = S3Backend::new(&config.s3).await?;
    let prefix = config.s3_prefix();

    // List all bases
    let base_prefix = format!("{}/base/", prefix);
    let mut bases: Vec<String> = backend
        .list_objects(&base_prefix)
        .await?
        .iter()
        .filter(|k| k.ends_with(".snap") || k.ends_with(".snap.gz"))
        .cloned()
        .collect();

    bases.sort();

    // Keep only the last N bases
    if bases.len() > config.retention.bases {
        let to_delete = bases.len() - config.retention.bases;
        eprintln!("Deleting {} old base snapshot(s)...", to_delete);

        for base in bases.iter().take(to_delete) {
            eprintln!("  Deleting: {}", base);
            backend.delete_object(base).await?;
        }
    }

    // Prune segments for deleted bases
    // This is simplified - in practice, you'd track which segments belong to which bases
    let segment_prefix = format!("{}/segments/", prefix);
    let segments = backend.list_objects(&segment_prefix).await?;

    // Note: Actually calculating sizes requires HEAD requests, so this is simplified
    // In a production implementation, you'd want to track segment sizes

    eprintln!("Found {} segments", segments.len());
    eprintln!("Pruning based on size limits not yet fully implemented");

    Ok(())
}