diff --git a/Cargo.toml b/Cargo.toml
index 3e6fb2e..bad9e46 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -4,3 +4,5 @@ version = "0.1.0"
 edition = "2024"
 
 [dependencies]
+chrono = "0.4.44"
+clap = { version = "4.6.1", features = ["derive"] }
diff --git a/src/command.rs b/src/command.rs
index 821e599..3d11f5a 100644
--- a/src/command.rs
+++ b/src/command.rs
@@ -2,7 +2,7 @@ pub fn exec_command(command: &Vec<&str>) -> Result<String, String> {
     if command.is_empty() {
         return Err("Command is empty".to_string());
     }
-    let mut cmd = std::process::Command::new(&command[0]);
+    let mut cmd = std::process::Command::new(command[0]);
     if command.len() > 1 {
         cmd.args(&command[1..]);
     }
@@ -18,3 +18,44 @@ pub fn exec_command(command: &Vec<&str>) -> Result<String, String> {
     let output_str = String::from_utf8_lossy(&output.stdout).to_string();
     Ok(output_str)
 }
+
+pub fn exec_piped_commands(source: &Vec<&str>, dest: &Vec<&str>) -> Result<(), String> {
+    if source.is_empty() || dest.is_empty() {
+        return Err("Source or destination command is empty".to_string());
+    }
+    let mut send_cmd = std::process::Command::new(source[0]);
+    if source.len() > 1 {
+        send_cmd.args(&source[1..]);
+    }
+    let mut receive_cmd = std::process::Command::new(dest[0]);
+    if dest.len() > 1 {
+        receive_cmd.args(&dest[1..]);
+    }
+
+    let mut send_process = send_cmd
+        .stdout(std::process::Stdio::piped())
+        .spawn()
+        .map_err(|e| e.to_string())?;
+
+    let mut receive_process = receive_cmd
+        .stdin(send_process.stdout.take().unwrap())
+        .spawn()
+        .map_err(|e| e.to_string())?;
+
+    let receive_status = receive_process.wait().map_err(|e| e.to_string())?;
+    let send_status = send_process.wait().map_err(|e| e.to_string())?;
+    if !receive_status.success() {
+        send_process.kill().ok();
+        return Err(format!(
+            "Receive {:?} failed with status {}",
+            dest, receive_status
+        ));
+    }
+    if !send_status.success() {
+        return Err(format!(
+            "Send command {:?} failed with status {}",
+            source, send_status
+        ));
+    }
+    Ok(())
+}
diff --git a/src/lib.rs b/src/lib.rs
index d69a522..033da8c 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -1,4 +1,6 @@
 mod command;
+use chrono::{Local, NaiveDateTime};
+use std::collections::HashSet;
 
 pub struct JobBuilder {
     sources: Vec<String>,
@@ -6,6 +8,7 @@ pub struct JobBuilder {
     source_zfs_command: Vec<String>,
     target_zfs_command: Vec<String>,
     dryrun: bool,
+    retain: usize,
 }
 
 fn parse_command(commandstr: &str) -> Vec<String> {
@@ -23,26 +26,24 @@ impl JobBuilder {
             source_zfs_command: vec!["zfs".to_string()],
             target_zfs_command: vec!["zfs".to_string()],
             dryrun: false,
+            retain: 2,
         }
     }
 
     pub fn source_zfs_command(mut self, commandstr: &str) -> Self {
-        let command: Vec<String>;
-        command = parse_command(commandstr);
+        let command = parse_command(commandstr);
         self.source_zfs_command = command;
         self
     }
 
     pub fn target_zfs_command(mut self, commandstr: &str) -> Self {
-        let command: Vec<String>;
-        command = parse_command(commandstr);
+        let command = parse_command(commandstr);
         self.target_zfs_command = command;
         self
     }
 
     pub fn zfs_command(mut self, commandstr: &str) -> Self {
-        let command: Vec<String>;
-        command = parse_command(commandstr);
+        let command = parse_command(commandstr);
         self.source_zfs_command = command.clone();
         self.target_zfs_command = command;
         self
@@ -53,6 +54,11 @@ impl JobBuilder {
         self
     }
 
+    pub fn retain(mut self, retain: usize) -> Self {
+        self.retain = retain;
+        self
+    }
+
     pub fn build(self) -> Result<Job, String> {
         let mut job = Job {
             datasets: vec![],
@@ -60,6 +66,7 @@ impl JobBuilder {
             source_zfs_command: self.source_zfs_command,
             target_zfs_command: self.target_zfs_command,
             dryrun: self.dryrun,
+            retain: self.retain,
         };
         let mut datasets: Vec<String> = vec![];
         for source in &self.sources {
@@ -75,7 +82,7 @@ impl JobBuilder {
             let mut cmd = job.get_side_command(JobSide::Source);
             cmd.extend(args);
             let output = command::exec_command(&cmd)?;
-            datasets = [datasets, output.lines().map(&str::to_string).collect()].concat();
+            datasets.extend(output.lines().map(str::to_string));
         }
         job.datasets = datasets;
         Ok(job)
@@ -88,13 +95,21 @@ pub struct Job {
     source_zfs_command: Vec<String>,
     target_zfs_command: Vec<String>,
     dryrun: bool,
+    retain: usize,
 }
 
+#[derive(Debug)]
 enum JobSide {
     Source,
     Destination,
 }
 
+#[derive(Debug)]
+struct Snapshot {
+    snapshot: String,
+    side: JobSide,
+}
+
 impl Job {
     pub fn dump(&self) {
         println!("Datasets: {:?}", self.datasets);
@@ -110,4 +125,158 @@ impl Job {
             JobSide::Destination => self.target_zfs_command.iter().map(|s| s.as_str()).collect(),
         }
     }
+
+    fn create_snapshot(&self, dataset: &str, side: JobSide) -> Result<String, String> {
+        let snapshot = format!("{}@{}", dataset, Local::now().format("%Y-%m-%dT%H:%M:%S"));
+        let mut cmd = self.get_side_command(side);
+        cmd.extend(["snapshot", &snapshot]);
+        command::exec_command(&cmd)?;
+        Ok(snapshot)
+    }
+
+    fn send_receive(
+        &self,
+        source: &str,
+        dest: &str,
+        inc_snapshot: Option<&str>,
+    ) -> Result<(), String> {
+        let mut send_cmd = self.get_side_command(JobSide::Source);
+        send_cmd.push("send");
+        if let Some(inc) = inc_snapshot {
+            send_cmd.extend(["-i", inc]);
+        }
+        send_cmd.push(source);
+
+        let mut receive_cmd = self.get_side_command(JobSide::Destination);
+        receive_cmd.extend(["receive", "-F", dest]);
+        command::exec_piped_commands(&send_cmd, &receive_cmd)?;
+        Ok(())
+    }
+
+    fn list_snapshots(&self, source: &str, side: JobSide) -> Result<Vec<String>, String> {
+        let mut cmd = self.get_side_command(side);
+        cmd.extend(["list", "-H", "-o", "name", "-t", "snapshot", source]);
+        let output = command::exec_command(&cmd)?;
+        let snapshots: Vec<&str> = output
+            .split_whitespace()
+            .map(|s| {
+                let parts: Vec<&str> = s.split("@").collect();
+                if parts.len() == 2 {
+                    Ok(parts[1])
+                } else {
+                    Err(format!("invalid snapshot name: {}", s))
+                }
+            })
+            .collect::<Result<Vec<_>, _>>()?;
+        let result: Vec<String> = snapshots
+            .iter()
+            .filter(|s| NaiveDateTime::parse_from_str(s, "%Y-%m-%dT%H:%M:%S").is_ok())
+            .map(|s| s.to_string())
+            .collect();
+        Ok(result)
+    }
+
+    fn find_latest_matching_snapshot(&self, source: &str, dest: &str) -> Result<String, String> {
+        self.find_matching_snapshots(source, dest)?
+            .into_iter()
+            .next()
+            .ok_or(String::from("no matching snapshots"))
+    }
+
+    fn find_matching_snapshots(&self, source: &str, dest: &str) -> Result<Vec<String>, String> {
+        let source_snapshots = self.list_snapshots(source, JobSide::Source)?;
+        let mut dest_snapshots = self.list_snapshots(dest, JobSide::Destination)?;
+        dest_snapshots.sort_by(|a, b| b.cmp(a));
+
+        let source_set: HashSet<String> = source_snapshots.into_iter().collect();
+
+        let matching_snapshots: Vec<String> = dest_snapshots
+            .into_iter()
+            .filter(|s| source_set.contains(s))
+            .collect();
+
+        if matching_snapshots.is_empty() {
+            Err(String::from("no matching snapshots"))
+        } else {
+            Ok(matching_snapshots)
+        }
+    }
+
+    fn find_old_snapshots(&self, source: &str, dest: &str) -> Result<Vec<Snapshot>, String> {
+        let source_snapshots = self.list_snapshots(source, JobSide::Source)?;
+        let dest_snapshots = self.list_snapshots(dest, JobSide::Destination)?;
+        let matching_snapshots = self.find_matching_snapshots(source, dest)?;
+        if matching_snapshots.is_empty() {
+            return Err(String::from("no matching snapshots found"));
+        }
+        let retain: HashSet<String> = matching_snapshots.into_iter().take(self.retain).collect();
+        let result = source_snapshots
+            .into_iter()
+            .filter(|s| !retain.contains(s))
+            .map(|s| Snapshot {
+                snapshot: format!("{}@{}", source, s),
+                side: JobSide::Source,
+            })
+            .chain(
+                dest_snapshots
+                    .into_iter()
+                    .filter(|s| !retain.contains(s))
+                    .map(|s| Snapshot {
+                        snapshot: format!("{}@{}", dest, s),
+                        side: JobSide::Destination,
+                    }),
+            )
+            .collect();
+        Ok(result)
+    }
+
+    fn delete_snapshot(&self, snapshot: Snapshot) -> Result<String, String> {
+        let mut cmd = self.get_side_command(snapshot.side);
+        cmd.extend(["destroy", &snapshot.snapshot]);
+        command::exec_command(&cmd)
+    }
+
+    pub fn run(&self) -> Result<(), String> {
+        for source in &self.datasets {
+            // Check the source exists
+            let mut cmd = self.get_side_command(JobSide::Source);
+            cmd.extend(["list", "-H", "-o", "name", source]);
+            let _ = command::exec_command(&cmd)?;
+
+            // Check whether the destination exists
+            // TODO: This will assume the destination doesn't exist if the
+            // command fails for any reason.
+            let dest = format!("{}/{}", self.target, source);
+            cmd = self.get_side_command(JobSide::Destination);
+            cmd.extend(["list", "-H", "-o", "name", &dest]);
+            let dest_exists = command::exec_command(&cmd).is_ok();
+
+            // Run backup
+            if dest_exists {
+                println!("Incremental backup {} -> {}", source, dest);
+                let inc_snapshot = format!(
+                    "{}@{}",
+                    source,
+                    self.find_latest_matching_snapshot(source, &dest)?
+                );
+                println!("Found matching snapshot: {}", inc_snapshot);
+
+                let snapshot = self.create_snapshot(source, JobSide::Source)?;
+                self.send_receive(&snapshot, &dest, Some(&inc_snapshot))?;
+            } else {
+                println!("Full backup {} -> {}", source, dest);
+                let snapshot = self.create_snapshot(source, JobSide::Source)?;
+                self.send_receive(&snapshot, &dest, None)?;
+            }
+
+            // Clean up snapshots
+            let old_snapshots = self.find_old_snapshots(source, &dest)?;
+            println!("Old snapshots: {:?}", old_snapshots);
+            old_snapshots
+                .into_iter()
+                .map(|s| self.delete_snapshot(s))
+                .collect::<Result<Vec<_>, _>>()?;
+        }
+        Ok(())
+    }
 }
diff --git a/src/main.rs b/src/main.rs
index 9116479..d3f2921 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -1,10 +1,44 @@
+use clap::Parser;
 use std::env::args;
 
 use zfsbackup::JobBuilder;
 
-fn main() {
-    let job = JobBuilder::new(args().skip(1).collect(), "backup".to_string())
-        .build()
-        .expect("asplode");
-    job.dump()
+#[derive(Parser)]
+#[command(version, about, long_about = None)]
+struct Args {
+    #[arg(short, default_value = "backup")]
+    target: String,
+
+    #[arg(short = 'T', long)]
+    target_zfs_command: Option<String>,
+
+    #[arg(short, long)]
+    source_zfs_command: Option<String>,
+
+    #[arg(short, long)]
+    zfs_command: Option<String>,
+
+    #[arg(short, long)]
+    dry_run: bool,
+
+    datasets: Vec<String>,
+}
+
+fn main() {
+    let args = Args::parse();
+    let mut builder = JobBuilder::new(args.datasets, args.target);
+    if args.dry_run {
+        builder = builder.dryrun();
+    }
+    if let Some(cmd) = args.zfs_command {
+        builder = builder.zfs_command(&cmd);
+    }
+    if let Some(cmd) = args.source_zfs_command {
+        builder = builder.source_zfs_command(&cmd);
+    }
+    if let Some(cmd) = args.target_zfs_command {
+        builder = builder.target_zfs_command(&cmd);
+    }
+    let job = builder.build().expect("asplode");
+    job.run().expect("boom");
 }