diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
new file mode 100644
index 0000000..968dcbc
--- /dev/null
+++ b/.github/workflows/ci.yml
@@ -0,0 +1,31 @@
+name: CI
+
+on:
+  pull_request:
+  push:
+    branches:
+      - main
+
+jobs:
+  test:
+    name: Run Rust tests
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Install Rust toolchain
+        uses: dtolnay/rust-toolchain@stable
+
+      - name: Cache Cargo dependencies
+        uses: Swatinem/rust-cache@v2
+
+      - name: Check formatting
+        run: cargo fmt --all -- --check
+
+      - name: Run Clippy
+        run: cargo clippy --all-targets --all-features -- -D warnings
+
+      - name: Run tests
+        run: cargo test --all
\ No newline at end of file
diff --git a/src/config.rs b/src/config.rs
index e5f7129..e4a6dc4 100644
--- a/src/config.rs
+++ b/src/config.rs
@@ -1,9 +1,19 @@
+use crate::filter::FilterRule;
+use serde::Deserialize;
 use std::path::PathBuf;
 
+#[derive(Debug, Deserialize, Clone)]
 pub struct Config {
     pub source_path: PathBuf,
     pub sink_url: String,
     pub batch_size: usize,
     pub flush_interval_secs: u64,
     pub channel_capacity: usize,
-}
\ No newline at end of file
+
+    #[serde(default = "filter_rules_default")]
+    pub filter_rules: Vec<FilterRule>,
+}
+
+fn filter_rules_default() -> Vec<FilterRule> {
+    vec![]
+}
diff --git a/src/filter.rs b/src/filter.rs
new file mode 100644
index 0000000..0ee86a0
--- /dev/null
+++ b/src/filter.rs
@@ -0,0 +1,56 @@
+use crate::record::LogLine;
+use regex::Regex;
+use serde::Deserialize;
+use tokio::sync::mpsc::{Receiver, Sender};
+
+#[derive(Debug, Deserialize, Clone)]
+pub enum RuleOp {
+    Equals,
+    Contains,
+    Regex,
+}
+
+#[derive(Debug, Deserialize, Clone)]
+pub enum RuleAction {
+    Drop,
+    Mask,
+}
+
+#[derive(Debug, Deserialize, Clone)]
+pub struct FilterRule {
+    pub field: String,
+    pub op: RuleOp,
+    pub value: String,
+    pub action: RuleAction,
+}
+
+pub async fn run_filter(mut rx: Receiver<LogLine>, tx: Sender<LogLine>, rules: Vec<FilterRule>) {
+    while let Some(mut log) = rx.recv().await {
+        let mut dropped = false;
+
+        for rule in &rules {
+            let field_val = log.get(&rule.field).and_then(|v| v.as_str()).unwrap_or("");
+            let matched = match rule.op {
+                RuleOp::Equals => field_val == rule.value,
+                RuleOp::Contains => field_val.contains(&rule.value),
+                RuleOp::Regex => Regex::new(&rule.value).unwrap().is_match(field_val),
+            };
+
+            if matched {
+                match rule.action {
+                    RuleAction::Drop => {
+                        dropped = true;
+                        break;
+                    }
+                    RuleAction::Mask => {
+                        log[&rule.field] = serde_json::json!("***");
+                    }
+                }
+            }
+        }
+
+        if !dropped && tx.send(log).await.is_err() {
+            break;
+        }
+    }
+}
diff --git a/src/lib.rs b/src/lib.rs
index 7531c95..ebadaed 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -1,2 +1,4 @@
+pub mod config;
+pub mod filter;
 pub mod record;
-pub mod config;
\ No newline at end of file
+pub mod sink;
diff --git a/src/record.rs b/src/record.rs
index a753002..f4f1a32 100644
--- a/src/record.rs
+++ b/src/record.rs
@@ -1 +1 @@
-pub type LogLine = serde_json::Value;
\ No newline at end of file
+pub type LogLine = serde_json::Value;
diff --git a/src/sink.rs b/src/sink.rs
new file mode 100644
index 0000000..bde31b6
--- /dev/null
+++ b/src/sink.rs
@@ -0,0 +1,46 @@
+use crate::config::Config;
+use crate::record::LogLine;
+use std::time::Duration;
+use tokio::sync::mpsc::Receiver;
+use tokio::time::interval;
+
+async fn flush(client: &reqwest::Client, url: &str, batch: &mut Vec<LogLine>) {
+    if batch.is_empty() {
+        return;
+    }
+
+    match client.post(url).json(batch).send().await {
+        Ok(resp) if resp.status().is_success() => {}
+        Ok(resp) => eprintln!(
+            "logtap: destination responded with status {}",
+            resp.status()
+        ),
+        Err(err) => eprintln!(
+            "logtap: failed to send batch ({} items): {err}",
+            batch.len()
+        ),
+    }
+
+    batch.clear();
+}
+pub async fn run_sink(cfg: Config, mut rx: Receiver<LogLine>) {
+    let client = reqwest::Client::new();
+    let mut batch: Vec<LogLine> = Vec::with_capacity(cfg.batch_size);
+    let mut ticker = interval(Duration::from_secs(cfg.flush_interval_secs));
+
+    loop {
+        tokio::select! {
+            Some(log) = rx.recv() => {
+                batch.push(log);
+                if batch.len() >= cfg.batch_size {
+                    flush(&client, &cfg.sink_url, &mut batch).await;
+                }
+            }
+            _ = ticker.tick() => {
+                if !batch.is_empty() {
+                    flush(&client, &cfg.sink_url, &mut batch).await;
+                }
+            }
+        }
+    }
+}
diff --git a/tests/filter_test.rs b/tests/filter_test.rs
new file mode 100644
index 0000000..748985d
--- /dev/null
+++ b/tests/filter_test.rs
@@ -0,0 +1,131 @@
+use logtap::filter::{FilterRule, RuleAction, RuleOp, run_filter};
+
+#[tokio::test]
+async fn filter_drops_log_when_equals_rule_matches() {
+    let (input_tx, input_rx) = tokio::sync::mpsc::channel(10);
+    let (output_tx, mut output_rx) = tokio::sync::mpsc::channel(10);
+
+    let rules = vec![FilterRule {
+        field: "level".to_string(),
+        op: RuleOp::Equals,
+        value: "debug".to_string(),
+        action: RuleAction::Drop,
+    }];
+
+    let filter = tokio::spawn(run_filter(input_rx, output_tx, rules));
+
+    input_tx
+        .send(serde_json::json!({
+            "level": "debug",
+            "message": "this should be dropped"
+        }))
+        .await
+        .unwrap();
+
+    drop(input_tx);
+
+    let received = output_rx.recv().await;
+
+    filter.await.unwrap();
+
+    assert!(received.is_none());
+}
+
+#[tokio::test]
+async fn filter_keeps_log_when_no_rule_matches() {
+    let (input_tx, input_rx) = tokio::sync::mpsc::channel(10);
+    let (output_tx, mut output_rx) = tokio::sync::mpsc::channel(10);
+
+    let rules = vec![FilterRule {
+        field: "level".to_string(),
+        op: RuleOp::Equals,
+        value: "debug".to_string(),
+        action: RuleAction::Drop,
+    }];
+
+    let filter = tokio::spawn(run_filter(input_rx, output_tx, rules));
+
+    input_tx
+        .send(serde_json::json!({
+            "level": "info",
+            "message": "this should pass"
+        }))
+        .await
+        .unwrap();
+
+    drop(input_tx);
+
+    let received = output_rx.recv().await.unwrap();
+
+    filter.await.unwrap();
+
+    assert_eq!(
+        received,
+        serde_json::json!({
+            "level": "info",
+            "message": "this should pass"
+        })
+    );
+}
+
+#[tokio::test]
+async fn filter_masks_field_when_contains_rule_matches() {
+    let (input_tx, input_rx) = tokio::sync::mpsc::channel(10);
+    let (output_tx, mut output_rx) = tokio::sync::mpsc::channel(10);
+
+    let rules = vec![FilterRule {
+        field: "email".to_string(),
+        op: RuleOp::Contains,
+        value: "@".to_string(),
+        action: RuleAction::Mask,
+    }];
+
+    let filter = tokio::spawn(run_filter(input_rx, output_tx, rules));
+
+    input_tx
+        .send(serde_json::json!({
+            "email": "user@example.com",
+            "message": "user logged in"
+        }))
+        .await
+        .unwrap();
+
+    drop(input_tx);
+
+    let received = output_rx.recv().await.unwrap();
+
+    filter.await.unwrap();
+
+    assert_eq!(received["email"], serde_json::json!("***"));
+    assert_eq!(received["message"], serde_json::json!("user logged in"));
+}
+
+#[tokio::test]
+async fn filter_applies_regex_rule() {
+    let (input_tx, input_rx) = tokio::sync::mpsc::channel(10);
+    let (output_tx, mut output_rx) = tokio::sync::mpsc::channel(10);
+
+    let rules = vec![FilterRule {
+        field: "message".to_string(),
+        op: RuleOp::Regex,
+        value: r"\d{3}-\d{2}-\d{4}".to_string(),
+        action: RuleAction::Mask,
+    }];
+
+    let filter = tokio::spawn(run_filter(input_rx, output_tx, rules));
+
+    input_tx
+        .send(serde_json::json!({
+            "message": "ssn 123-45-6789"
+        }))
+        .await
+        .unwrap();
+
+    drop(input_tx);
+
+    let received = output_rx.recv().await.unwrap();
+
+    filter.await.unwrap();
+
+    assert_eq!(received["message"], serde_json::json!("***"));
+}
diff --git a/tests/sink_test.rs b/tests/sink_test.rs
new file mode 100644
index 0000000..72dc400
--- /dev/null
+++ b/tests/sink_test.rs
@@ -0,0 +1,105 @@
+use std::path::PathBuf;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::TcpListener;
+
+use logtap::config::Config;
+use logtap::record::LogLine;
+use logtap::sink::run_sink;
+
+async fn read_http_request_body(stream: &mut tokio::net::TcpStream) -> String {
+    let mut buffer = Vec::new();
+    let mut temp = [0_u8; 1024];
+
+    loop {
+        let read = stream.read(&mut temp).await.unwrap();
+        assert!(read > 0, "connection closed before headers were read");
+
+        buffer.extend_from_slice(&temp[..read]);
+
+        if buffer.windows(4).any(|window| window == b"\r\n\r\n") {
+            break;
+        }
+    }
+
+    let headers_end = buffer
+        .windows(4)
+        .position(|window| window == b"\r\n\r\n")
+        .unwrap()
+        + 4;
+
+    let headers = String::from_utf8_lossy(&buffer[..headers_end]);
+    let content_length = headers
+        .lines()
+        .find_map(|line| {
+            line.to_ascii_lowercase()
+                .strip_prefix("content-length:")
+                .map(|value| value.trim().parse::<usize>().unwrap())
+        })
+        .unwrap();
+
+    while buffer.len() < headers_end + content_length {
+        let read = stream.read(&mut temp).await.unwrap();
+        assert!(read > 0, "connection closed before body was read");
+
+        buffer.extend_from_slice(&temp[..read]);
+    }
+
+    String::from_utf8(buffer[headers_end..headers_end + content_length].to_vec()).unwrap()
+}
+
+#[tokio::test]
+async fn sink_posts_logs_when_batch_size_is_reached() {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let address = listener.local_addr().unwrap();
+
+    let server = tokio::spawn(async move {
+        let (mut stream, _) = listener.accept().await.unwrap();
+
+        let body = read_http_request_body(&mut stream).await;
+
+        stream
+            .write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 0\r\nconnection: close\r\n\r\n")
+            .await
+            .unwrap();
+
+        body
+    });
+
+    let cfg = Config {
+        source_path: PathBuf::from("unused.log"),
+        sink_url: format!("http://{address}/logs"),
+        batch_size: 2,
+        flush_interval_secs: 60,
+        channel_capacity: 10,
+        filter_rules: vec![],
+    };
+
+    let (tx, rx) = tokio::sync::mpsc::channel::<LogLine>(cfg.channel_capacity);
+
+    let sink = tokio::spawn(run_sink(cfg, rx));
+
+    tx.send(serde_json::json!({ "message": "first log" }))
+        .await
+        .unwrap();
+
+    tx.send(serde_json::json!({ "message": "second log" }))
+        .await
+        .unwrap();
+
+    let body = tokio::time::timeout(std::time::Duration::from_secs(2), server)
+        .await
+        .unwrap()
+        .unwrap();
+
+    sink.abort();
+
+    let received_logs: serde_json::Value = serde_json::from_str(&body).unwrap();
+
+    assert_eq!(
+        received_logs,
+        serde_json::json!([
+            { "message": "first log" },
+            { "message": "second log" }
+        ])
+    );
+}