From b8e6caaa41fc191fd1ef7484789aca24749110ad Mon Sep 17 00:00:00 2001
From: Trezy <tre@trezy.com>
Date: Sun, 15 Mar 2026 09:20:06 -0500
Subject: [PATCH] feat: add support for labelers

---
 ...314100000_create_labeler_subscriptions.sql |   7 +
 migrations/20260314100001_create_labels.sql   |  11 +
 src/admin/labelers.rs                         | 150 ++++++
 src/admin/mod.rs                              |   6 +
 src/admin/permissions.rs                      |  20 +-
 src/admin/types.rs                            |  23 +
 src/aip.rs                                    |   2 +
 src/labeler.rs                                | 440 ++++++++++++++++++
 src/lib.rs                                    |   2 +
 src/lua/atproto_api.rs                        | 144 ++++++
 src/lua/db_api.rs                             |   2 +
 src/lua/execute.rs                            |   2 +
 src/lua/http_api.rs                           |   2 +
 src/main.rs                                   |   7 +-
 src/tap.rs                                    |   2 +
 tests/common/app.rs                           |   2 +
 tests/common/db.rs                            |   2 +-
 tests/e2e_labelers.rs                         | 387 +++++++++++++++
 tests/lua_atproto_api.rs                      | 386 +++++++++++++++
 tests/lua_db_api.rs                           |   2 +
 20 files changed, 1595 insertions(+), 4 deletions(-)
 create mode 100644 migrations/20260314100000_create_labeler_subscriptions.sql
 create mode 100644 migrations/20260314100001_create_labels.sql
 create mode 100644 src/admin/labelers.rs
 create mode 100644 src/labeler.rs
 create mode 100644 tests/e2e_labelers.rs
 create mode 100644 tests/lua_atproto_api.rs

diff --git a/migrations/20260314100000_create_labeler_subscriptions.sql b/migrations/20260314100000_create_labeler_subscriptions.sql
new file mode 100644
index 0000000..7f40edf
--- /dev/null
+++ b/migrations/20260314100000_create_labeler_subscriptions.sql
@@ -0,0 +1,7 @@
+CREATE TABLE labeler_subscriptions (
+    did        TEXT PRIMARY KEY,
+    cursor     BIGINT,
+    status     TEXT NOT NULL DEFAULT 'active',
+    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
diff --git a/migrations/20260314100001_create_labels.sql b/migrations/20260314100001_create_labels.sql
new file mode 100644
index 0000000..43dbd50
--- /dev/null
+++ b/migrations/20260314100001_create_labels.sql
@@ -0,0 +1,11 @@
+CREATE TABLE labels (
+    src TEXT NOT NULL,
+    uri TEXT NOT NULL,
+    val TEXT NOT NULL,
+    cts TIMESTAMPTZ NOT NULL,
+    exp TIMESTAMPTZ,
+    PRIMARY KEY (src, uri, val)
+);
+
+CREATE INDEX idx_labels_uri ON labels (uri);
+CREATE INDEX idx_labels_exp ON labels (exp) WHERE exp IS NOT NULL;
diff --git a/src/admin/labelers.rs b/src/admin/labelers.rs
new file mode 100644
index 0000000..a7f8c78
--- /dev/null
+++ b/src/admin/labelers.rs
@@ -0,0 +1,150 @@
+use axum::Json;
+use axum::extract::{Path, State};
+use axum::http::StatusCode;
+
+use crate::AppState;
+use crate::error::AppError;
+use crate::event_log::{EventLog, Severity, log_event};
+
+use super::auth::UserAuth;
+use super::permissions::Permission;
+use super::types::{AddLabelerBody, LabelerSummary, UpdateLabelerBody};
+
+/// GET /admin/labelers — list all labeler subscriptions.
+pub(super) async fn list(
+    State(state): State<AppState>,
+    auth: UserAuth,
+) -> Result<Json<Vec<LabelerSummary>>, AppError> {
+    auth.require(Permission::LabelersRead).await?;
+
+    let labelers: Vec<LabelerSummary> = sqlx::query_as(
+        "SELECT did, status, cursor, created_at, updated_at FROM labeler_subscriptions ORDER BY created_at",
+    )
+    .fetch_all(&state.db)
+    .await
+    .map_err(|e| AppError::Internal(format!("failed to list labeler subscriptions: {e}")))?;
+
+    Ok(Json(labelers))
+}
+
+/// POST /admin/labelers — add a labeler subscription.
+pub(super) async fn add(
+    State(state): State<AppState>,
+    auth: UserAuth,
+    Json(body): Json<AddLabelerBody>,
+) -> Result<StatusCode, AppError> {
+    auth.require(Permission::LabelersCreate).await?;
+
+    sqlx::query(
+        r#"
+        INSERT INTO labeler_subscriptions (did)
+        VALUES ($1)
+        ON CONFLICT (did) DO UPDATE SET status = 'active', updated_at = NOW()
+        "#,
+    )
+    .bind(&body.did)
+    .execute(&state.db)
+    .await
+    .map_err(|e| AppError::Internal(format!("failed to add labeler subscription: {e}")))?;
+
+    // Notify the labeler consumer to pick up the new subscription.
+    let _ = state.labeler_subscriptions_tx.send(());
+
+    log_event(
+        &state.db,
+        EventLog {
+            event_type: "labeler.added".to_string(),
+            severity: Severity::Info,
+            actor_did: Some(auth.did.clone()),
+            subject: Some(body.did.clone()),
+            detail: serde_json::json!({}),
+        },
+    )
+    .await;
+
+    Ok(StatusCode::CREATED)
+}
+
+/// PATCH /admin/labelers/{did} — update labeler status (active/paused).
+pub(super) async fn update(
+    State(state): State<AppState>,
+    auth: UserAuth,
+    Path(did): Path<String>,
+    Json(body): Json<UpdateLabelerBody>,
+) -> Result<StatusCode, AppError> {
+    auth.require(Permission::LabelersCreate).await?;
+
+    let result = sqlx::query(
+        "UPDATE labeler_subscriptions SET status = $1, updated_at = NOW() WHERE did = $2",
+    )
+    .bind(&body.status)
+    .bind(&did)
+    .execute(&state.db)
+    .await
+    .map_err(|e| AppError::Internal(format!("failed to update labeler subscription: {e}")))?;
+
+    if result.rows_affected() == 0 {
+        return Err(AppError::NotFound(format!(
+            "labeler subscription '{did}' not found"
+        )));
+    }
+
+    let _ = state.labeler_subscriptions_tx.send(());
+
+    log_event(
+        &state.db,
+        EventLog {
+            event_type: "labeler.updated".to_string(),
+            severity: Severity::Info,
+            actor_did: Some(auth.did.clone()),
+            subject: Some(did),
+            detail: serde_json::json!({ "status": body.status }),
+        },
+    )
+    .await;
+
+    Ok(StatusCode::NO_CONTENT)
+}
+
+/// DELETE /admin/labelers/{did} — remove a labeler subscription and its labels.
+pub(super) async fn delete(
+    State(state): State<AppState>,
+    auth: UserAuth,
+    Path(did): Path<String>,
+) -> Result<StatusCode, AppError> {
+    auth.require(Permission::LabelersDelete).await?;
+
+    let result = sqlx::query("DELETE FROM labeler_subscriptions WHERE did = $1")
+        .bind(&did)
+        .execute(&state.db)
+        .await
+        .map_err(|e| AppError::Internal(format!("failed to delete labeler subscription: {e}")))?;
+
+    if result.rows_affected() == 0 {
+        return Err(AppError::NotFound(format!(
+            "labeler subscription '{did}' not found"
+        )));
+    }
+
+    // Also remove all labels from this labeler.
+    let _ = sqlx::query("DELETE FROM labels WHERE src = $1")
+        .bind(&did)
+        .execute(&state.db)
+        .await;
+
+    let _ = state.labeler_subscriptions_tx.send(());
+
+    log_event(
+        &state.db,
+        EventLog {
+            event_type: "labeler.deleted".to_string(),
+            severity: Severity::Info,
+            actor_did: Some(auth.did.clone()),
+            subject: Some(did),
+            detail: serde_json::json!({}),
+        },
+    )
+    .await;
+
+    Ok(StatusCode::NO_CONTENT)
+}
diff --git a/src/admin/mod.rs b/src/admin/mod.rs
index 34dd5dc..2eef7eb 100644
--- a/src/admin/mod.rs
+++ b/src/admin/mod.rs
@@ -2,6 +2,7 @@ mod api_keys;
 pub(crate) mod auth;
 mod backfill;
 mod events;
+mod labelers;
 mod lexicons;
 mod network_lexicons;
 pub(crate) mod permissions;
@@ -62,4 +63,9 @@ pub fn admin_routes(_state: AppState) -> Router<AppState> {
             post(script_variables::upsert).get(script_variables::list),
         )
         .route("/script-variables/{key}", delete(script_variables::delete))
+        .route("/labelers", post(labelers::add).get(labelers::list))
+        .route(
+            "/labelers/{did}",
+            patch(labelers::update).delete(labelers::delete),
+        )
 }
diff --git a/src/admin/permissions.rs b/src/admin/permissions.rs
index cb09a57..ee0edf1 100644
--- a/src/admin/permissions.rs
+++ b/src/admin/permissions.rs
@@ -2,7 +2,7 @@ use std::collections::HashSet;
 
 use serde::{Deserialize, Serialize};
 
-/// All 20 permissions in the system.
+/// All 23 permissions in the system.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
 pub enum Permission {
     #[serde(rename = "lexicons:create")]
@@ -52,6 +52,13 @@ pub enum Permission {
 
     #[serde(rename = "events:read")]
     EventsRead,
+
+    #[serde(rename = "labelers:create")]
+    LabelersCreate,
+    #[serde(rename = "labelers:read")]
+    LabelersRead,
+    #[serde(rename = "labelers:delete")]
+    LabelersDelete,
 }
 
 impl Permission {
@@ -78,10 +85,13 @@ impl Permission {
             Self::BackfillRead => "backfill:read",
             Self::StatsRead => "stats:read",
             Self::EventsRead => "events:read",
+            Self::LabelersCreate => "labelers:create",
+            Self::LabelersRead => "labelers:read",
+            Self::LabelersDelete => "labelers:delete",
         }
     }
 
-    /// All 20 permissions.
+    /// All 23 permissions.
     pub fn all() -> HashSet<Permission> {
         HashSet::from([
             Self::LexiconsCreate,
@@ -104,6 +114,9 @@ impl Permission {
             Self::BackfillRead,
             Self::StatsRead,
             Self::EventsRead,
+            Self::LabelersCreate,
+            Self::LabelersRead,
+            Self::LabelersDelete,
         ])
     }
 }
@@ -145,6 +158,9 @@ impl Template {
                 perms.insert(Permission::ScriptVariablesCreate);
                 perms.insert(Permission::ScriptVariablesDelete);
                 perms.insert(Permission::RecordsDelete);
+                perms.insert(Permission::LabelersCreate);
+                perms.insert(Permission::LabelersRead);
+                perms.insert(Permission::LabelersDelete);
                 perms
             }
             Self::FullAccess => Permission::all(),
diff --git a/src/admin/types.rs b/src/admin/types.rs
index 87b1a33..6026d67 100644
--- a/src/admin/types.rs
+++ b/src/admin/types.rs
@@ -181,3 +181,26 @@ pub(super) struct UpsertScriptVariableBody {
     pub(super) key: String,
     pub(super) value: String,
 }
+
+// ---------------------------------------------------------------------------
+// Labeler subscription types
+// ---------------------------------------------------------------------------
+
+#[derive(Deserialize)]
+pub(super) struct AddLabelerBody {
+    pub(super) did: String,
+}
+
+#[derive(Serialize, sqlx::FromRow)]
+pub(super) struct LabelerSummary {
+    pub(super) did: String,
+    pub(super) status: String,
+    pub(super) cursor: Option<i64>,
+    pub(super) created_at: chrono::DateTime<chrono::Utc>,
+    pub(super) updated_at: chrono::DateTime<chrono::Utc>,
+}
+
+#[derive(Deserialize)]
+pub(super) struct UpdateLabelerBody {
+    pub(super) status: String,
+}
diff --git a/src/aip.rs b/src/aip.rs
index 7941edd..84c7b7f 100644
--- a/src/aip.rs
+++ b/src/aip.rs
@@ -111,12 +111,14 @@ mod tests {
             event_log_retention_days: 30,
         };
         let (tx, _) = watch::channel(vec![]);
+        let (labeler_tx, _) = watch::channel(());
         AppState {
             config,
             http: reqwest::Client::new(),
             db: sqlx::PgPool::connect_lazy("postgres://localhost/fake").unwrap(),
             lexicons: crate::lexicon::LexiconRegistry::new(),
             collections_tx: tx,
+            labeler_subscriptions_tx: labeler_tx,
         }
     }
 
diff --git a/src/labeler.rs b/src/labeler.rs
new file mode 100644
index 0000000..1451db0
--- /dev/null
+++ b/src/labeler.rs
@@ -0,0 +1,440 @@
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use futures_util::StreamExt;
+use serde::Deserialize;
+use tokio::sync::watch;
+use tokio::task::JoinHandle;
+use tokio_tungstenite::tungstenite::Message;
+use tokio_tungstenite::tungstenite::client::IntoClientRequest;
+
+use crate::AppState;
+use crate::event_log::{EventLog, Severity, log_event};
+use crate::profile;
+
+// ---------------------------------------------------------------------------
+// Types
+// ---------------------------------------------------------------------------
+
+#[derive(Deserialize)]
+struct SubscribeLabelsMessage {
+    seq: i64,
+    labels: Vec<Label>,
+}
+
+#[derive(Deserialize)]
+struct Label {
+    src: String,
+    uri: String,
+    val: String,
+    #[serde(default)]
+    neg: bool,
+    cts: String,
+    exp: Option<String>,
+}
+
+// Used for the queryLabels HTTP response.
+#[derive(Deserialize)]
+struct QueryLabelsResponse {
+    labels: Vec<Label>,
+}
+
+// ---------------------------------------------------------------------------
+// Spawn — manages per-labeler subscription tasks
+// ---------------------------------------------------------------------------
+
+pub fn spawn(state: AppState, mut subscriptions_rx: watch::Receiver<()>) {
+    tokio::spawn(async move {
+        let mut tasks: HashMap<String, JoinHandle<()>> = HashMap::new();
+
+        loop {
+            // Read all active subscriptions from the database.
+            let active: Vec<(String,)> =
+                sqlx::query_as("SELECT did FROM labeler_subscriptions WHERE status = 'active'")
+                    .fetch_all(&state.db)
+                    .await
+                    .unwrap_or_default();
+
+            let active_dids: Vec<String> = active.into_iter().map(|(did,)| did).collect();
+
+            // Stop tasks for removed/paused subscriptions.
+            let to_remove: Vec<String> = tasks
+                .keys()
+                .filter(|did| !active_dids.contains(did))
+                .cloned()
+                .collect();
+
+            for did in to_remove {
+                if let Some(handle) = tasks.remove(&did) {
+                    tracing::info!(did = %did, "stopping labeler subscription");
+                    handle.abort();
+                }
+            }
+
+            // Start tasks for new subscriptions.
+            for did in &active_dids {
+                if !tasks.contains_key(did) {
+                    tracing::info!(did = %did, "starting labeler subscription");
+                    let state = state.clone();
+                    let did_clone = did.clone();
+                    let handle = tokio::spawn(run_subscription(state, did_clone));
+                    tasks.insert(did.clone(), handle);
+                }
+            }
+
+            // Wait for a signal that subscriptions have changed.
+            if subscriptions_rx.changed().await.is_err() {
+                tracing::info!("labeler subscriptions channel closed, stopping");
+                break;
+            }
+        }
+
+        // Clean up all tasks on exit.
+        for (_, handle) in tasks {
+            handle.abort();
+        }
+    });
+}
+
+// ---------------------------------------------------------------------------
+// Per-labeler reconnect loop
+// ---------------------------------------------------------------------------
+
+async fn run_subscription(state: AppState, did: String) {
+    loop {
+        match run_subscription_once(&state, &did).await {
+            Ok(()) => {
+                tracing::info!(did = %did, "labeler subscription ended cleanly, reconnecting");
+            }
+            Err(e) => {
+                tracing::warn!(did = %did, "labeler subscription error: {e}");
+            }
+        }
+        tokio::time::sleep(std::time::Duration::from_secs(2)).await;
+        tracing::info!(did = %did, "reconnecting to labeler");
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Single connection lifecycle
+// ---------------------------------------------------------------------------
+
+async fn run_subscription_once(
+    state: &AppState,
+    did: &str,
+) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    // Resolve the labeler's PDS endpoint from its DID.
+    let pds_endpoint = profile::resolve_pds_endpoint(&state.http, &state.config.plc_url, did)
+        .await
+        .map_err(|e| format!("failed to resolve PDS for {did}: {e:?}"))?;
+
+    // Convert HTTP URL to WebSocket URL.
+    let ws_url = http_to_ws(&pds_endpoint);
+
+    // Read cursor from database.
+    let cursor: Option<(Option<i64>,)> =
+        sqlx::query_as("SELECT cursor FROM labeler_subscriptions WHERE did = $1")
+            .bind(did)
+            .fetch_optional(&state.db)
+            .await?;
+
+    let cursor_val = cursor.and_then(|(c,)| c).unwrap_or(0);
+
+    let url = format!(
+        "{}/xrpc/com.atproto.label.subscribeLabels?cursor={}",
+        ws_url.trim_end_matches('/'),
+        cursor_val
+    );
+
+    tracing::info!(did = %did, url = %url, "connecting to labeler");
+
+    let request = url.into_client_request()?;
+
+    let (ws, _): (
+        tokio_tungstenite::WebSocketStream<
+            tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
+        >,
+        _,
+    ) = tokio_tungstenite::connect_async(request).await?;
+
+    tracing::info!(did = %did, "connected to labeler");
+
+    log_event(
+        &state.db,
+        EventLog {
+            event_type: "labeler.connected".to_string(),
+            severity: Severity::Info,
+            actor_did: None,
+            subject: Some(did.to_string()),
+            detail: serde_json::json!({ "did": did }),
+        },
+    )
+    .await;
+
+    let (_write, mut read) = ws.split();
+    let mut events_since_cursor_save: u64 = 0;
+    let mut last_seq: i64 = cursor_val;
+
+    loop {
+        let msg = match read.next().await {
+            Some(Ok(m)) => m,
+            Some(Err(e)) => {
+                tracing::warn!(did = %did, "labeler websocket read error: {e}");
+                // Persist cursor before disconnecting.
+                persist_cursor(&state.db, did, last_seq).await;
+                return Err(e.into());
+            }
+            None => {
+                tracing::info!(did = %did, "labeler websocket stream ended");
+                break;
+            }
+        };
+
+        let bytes = match msg {
+            Message::Text(t) => t.as_bytes().to_vec(),
+            Message::Binary(b) => b.to_vec(),
+            Message::Close(_) => {
+                tracing::info!(did = %did, "labeler websocket received close frame");
+                break;
+            }
+            _ => continue,
+        };
+
+        let message: SubscribeLabelsMessage = match serde_json::from_slice(&bytes) {
+            Ok(m) => m,
+            Err(e) => {
+                tracing::warn!(did = %did, "skipping unparseable labeler message: {e}");
+                continue;
+            }
+        };
+
+        last_seq = message.seq;
+
+        for label in &message.labels {
+            apply_label(&state.db, label).await;
+        }
+
+        events_since_cursor_save += 1;
+        if events_since_cursor_save >= 100 {
+            persist_cursor(&state.db, did, last_seq).await;
+            events_since_cursor_save = 0;
+        }
+    }
+
+    // Persist final cursor on disconnect.
+    persist_cursor(&state.db, did, last_seq).await;
+
+    log_event(
+        &state.db,
+        EventLog {
+            event_type: "labeler.disconnected".to_string(),
+            severity: Severity::Warn,
+            actor_did: None,
+            subject: Some(did.to_string()),
+            detail: serde_json::json!({ "did": did, "last_seq": last_seq }),
+        },
+    )
+    .await;
+
+    Ok(())
+}
+
+// ---------------------------------------------------------------------------
+// Helpers
+// ---------------------------------------------------------------------------
+
+fn http_to_ws(url: &str) -> String {
+    let base = url.trim_end_matches('/');
+    if let Some(rest) = base.strip_prefix("https://") {
+        format!("wss://{rest}")
+    } else if let Some(rest) = base.strip_prefix("http://") {
+        format!("ws://{rest}")
+    } else {
+        format!("ws://{base}")
+    }
+}
+
+async fn apply_label(db: &sqlx::PgPool, label: &Label) {
+    if label.neg {
+        // Negation label — remove it.
+        if let Err(e) = sqlx::query("DELETE FROM labels WHERE src = $1 AND uri = $2 AND val = $3")
+            .bind(&label.src)
+            .bind(&label.uri)
+            .bind(&label.val)
+            .execute(db)
+            .await
+        {
+            tracing::warn!(
+                src = %label.src, uri = %label.uri, val = %label.val,
+                "failed to delete negated label: {e}"
+            );
+        }
+    } else {
+        // Normal label — upsert.
+        let cts = chrono::DateTime::parse_from_rfc3339(&label.cts)
+            .map(|dt| dt.with_timezone(&chrono::Utc))
+            .ok();
+
+        let exp = label
+            .exp
+            .as_deref()
+            .and_then(|e| chrono::DateTime::parse_from_rfc3339(e).ok())
+            .map(|dt| dt.with_timezone(&chrono::Utc));
+
+        if let Err(e) = sqlx::query(
+            r#"
+            INSERT INTO labels (src, uri, val, cts, exp)
+            VALUES ($1, $2, $3, $4, $5)
+            ON CONFLICT (src, uri, val) DO UPDATE
+                SET cts = EXCLUDED.cts,
+                    exp = EXCLUDED.exp
+            "#,
+        )
+        .bind(&label.src)
+        .bind(&label.uri)
+        .bind(&label.val)
+        .bind(cts)
+        .bind(exp)
+        .execute(db)
+        .await
+        {
+            tracing::warn!(
+                src = %label.src, uri = %label.uri, val = %label.val,
+                "failed to upsert label: {e}"
+            );
+        }
+    }
+}
+
+async fn persist_cursor(db: &sqlx::PgPool, did: &str, seq: i64) {
+    if let Err(e) = sqlx::query(
+        "UPDATE labeler_subscriptions SET cursor = $1, updated_at = NOW() WHERE did = $2",
+    )
+    .bind(seq)
+    .bind(did)
+    .execute(db)
+    .await
+    {
+        tracing::warn!(did = %did, seq, "failed to persist labeler cursor: {e}");
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Backfill labels for a specific URI
+// ---------------------------------------------------------------------------
+
+/// Spawn a background task to backfill labels for a given URI from all active
+/// labeler subscriptions. Fire-and-forget.
+pub fn backfill_labels_for_uri(state: Arc<AppState>, uri: String) {
+    tokio::spawn(async move {
+        if let Err(e) = backfill_labels_for_uri_inner(&state, &uri).await {
+            tracing::warn!(uri = %uri, "failed to backfill labels: {e}");
+        }
+    });
+}
+
+async fn backfill_labels_for_uri_inner(
+    state: &AppState,
+    uri: &str,
+) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    let subscriptions: Vec<(String,)> =
+        sqlx::query_as("SELECT did FROM labeler_subscriptions WHERE status = 'active'")
+            .fetch_all(&state.db)
+            .await?;
+
+    for (labeler_did,) in subscriptions {
+        if let Err(e) = backfill_from_labeler(state, &labeler_did, uri).await {
+            tracing::warn!(
+                labeler = %labeler_did, uri = %uri,
+                "failed to backfill labels from labeler: {e}"
+            );
+        }
+    }
+
+    Ok(())
+}
+
+async fn backfill_from_labeler(
+    state: &AppState,
+    labeler_did: &str,
+    uri: &str,
+) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    let pds_endpoint =
+        profile::resolve_pds_endpoint(&state.http, &state.config.plc_url, labeler_did)
+            .await
+            .map_err(|e| format!("failed to resolve PDS for {labeler_did}: {e:?}"))?;
+
+    let encoded_uri = urlencoding::encode(uri);
+    let url = format!(
+        "{}/xrpc/com.atproto.label.queryLabels?uriPatterns={}",
+        pds_endpoint.trim_end_matches('/'),
+        encoded_uri
+    );
+
+    let resp = state.http.get(&url).send().await?;
+
+    if !resp.status().is_success() {
+        let status = resp.status();
+        let body = resp.text().await.unwrap_or_default();
+        return Err(format!("queryLabels returned {status}: {body}").into());
+    }
+
+    let response: QueryLabelsResponse = resp.json().await?;
+
+    for label in &response.labels {
+        apply_label(&state.db, label).await;
+    }
+
+    Ok(())
+}
+
+// ---------------------------------------------------------------------------
+// Label garbage collection
+// ---------------------------------------------------------------------------
+
+/// Hourly task to clean up expired and orphaned labels.
+pub async fn spawn_label_gc(db: sqlx::PgPool) {
+    tracing::info!("starting label garbage collection task");
+
+    let interval = tokio::time::Duration::from_secs(3600); // 1 hour
+    loop {
+        tokio::time::sleep(interval).await;
+
+        // Delete expired labels.
+        let expired = sqlx::query("DELETE FROM labels WHERE exp IS NOT NULL AND exp < NOW()")
+            .execute(&db)
+            .await;
+
+        let expired_count = match expired {
+            Ok(r) => r.rows_affected(),
+            Err(e) => {
+                tracing::warn!("failed to clean up expired labels: {e}");
+                0
+            }
+        };
+
+        // Delete orphaned labels (no matching record).
+        let orphaned = sqlx::query(
+            "DELETE FROM labels WHERE NOT EXISTS (SELECT 1 FROM records WHERE records.uri = labels.uri)",
+        )
+        .execute(&db)
+        .await;
+
+        let orphaned_count = match orphaned {
+            Ok(r) => r.rows_affected(),
+            Err(e) => {
+                tracing::warn!("failed to clean up orphaned labels: {e}");
+                0
+            }
+        };
+
+        let total = expired_count + orphaned_count;
+        if total > 0 {
+            tracing::info!(
+                expired = expired_count,
+                orphaned = orphaned_count,
+                "cleaned up {total} labels"
+            );
+        }
+    }
+}
diff --git a/src/lib.rs b/src/lib.rs
index 88a9d33..68ee8e6 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -4,6 +4,7 @@ pub mod auth;
 pub mod config;
 pub mod error;
 pub mod event_log;
+pub mod labeler;
 pub mod lexicon;
 pub mod lua;
 pub mod profile;
@@ -25,4 +26,5 @@ pub struct AppState {
     pub db: sqlx::PgPool,
     pub lexicons: LexiconRegistry,
     pub collections_tx: watch::Sender<Vec<String>>,
+    pub labeler_subscriptions_tx: watch::Sender<()>,
 }
diff --git a/src/lua/atproto_api.rs b/src/lua/atproto_api.rs
index ffbcced..18c3a00 100644
--- a/src/lua/atproto_api.rs
+++ b/src/lua/atproto_api.rs
@@ -23,6 +23,148 @@ pub fn register_atproto_api(lua: &Lua, state: Arc<AppState>) -> LuaResult<()> {
     })?;
 
     atproto_table.set("resolve_service_endpoint", resolve_fn)?;
+
+    // get_labels(uri) -> array of { src, uri, val, cts }
+    let state_clone = state.clone();
+    let get_labels_fn = lua.create_async_function(move |lua, uri: String| {
+        let state = state_clone.clone();
+        async move {
+            // Query external labels from the labels table.
+            let rows: Vec<(String, String, String, chrono::DateTime<chrono::Utc>)> =
+                sqlx::query_as(
+                    "SELECT src, uri, val, cts FROM labels WHERE uri = $1 AND (exp IS NULL OR exp > NOW())",
+                )
+                .bind(&uri)
+                .fetch_all(&state.db)
+                .await
+                .map_err(|e| mlua::Error::runtime(format!("label query failed: {e}")))?;
+
+            let result = lua.create_table()?;
+            let mut idx = 1;
+
+            for (src, label_uri, val, cts) in &rows {
+                let label = lua.create_table()?;
+                label.set("src", src.as_str())?;
+                label.set("uri", label_uri.as_str())?;
+                label.set("val", val.as_str())?;
+                label.set("cts", cts.to_rfc3339())?;
+                result.set(idx, label)?;
+                idx += 1;
+            }
+
+            // Check for self-labels in the record itself.
+            let record: Option<(String, serde_json::Value)> = sqlx::query_as(
+                "SELECT did, record FROM records WHERE uri = $1",
+            )
+            .bind(&uri)
+            .fetch_optional(&state.db)
+            .await
+            .map_err(|e| mlua::Error::runtime(format!("record query failed: {e}")))?;
+
+            if let Some((did, record)) = record
+                && let Some(labels) = record.get("labels")
+                && let Some(values) = labels.get("values")
+                && let Some(arr) = values.as_array()
+            {
+                for item in arr {
+                    if let Some(val) = item.get("val").and_then(|v| v.as_str()) {
+                        let label = lua.create_table()?;
+                        label.set("src", did.as_str())?;
+                        label.set("uri", uri.as_str())?;
+                        label.set("val", val)?;
+                        label.set("cts", "")?;
+                        result.set(idx, label)?;
+                        idx += 1;
+                    }
+                }
+            }
+
+            Ok(mlua::Value::Table(result))
+        }
+    })?;
+    atproto_table.set("get_labels", get_labels_fn)?;
+
+    // get_labels_batch(uris) -> table keyed by URI
+    let state_clone = state.clone();
+    let get_labels_batch_fn = lua.create_async_function(move |lua, uris: mlua::Table| {
+        let state = state_clone.clone();
+        async move {
+            // Collect URIs from the Lua table.
+            let uri_list: Vec<String> = uris
+                .sequence_values::<String>()
+                .collect::<Result<Vec<_>, _>>()?;
+
+            // Query all labels for all URIs at once.
+            let rows: Vec<(String, String, String, chrono::DateTime<chrono::Utc>)> =
+                sqlx::query_as(
+                    "SELECT src, uri, val, cts FROM labels WHERE uri = ANY($1) AND (exp IS NULL OR exp > NOW())",
+                )
+                .bind(&uri_list)
+                .fetch_all(&state.db)
+                .await
+                .map_err(|e| mlua::Error::runtime(format!("label batch query failed: {e}")))?;
+
+            // Query records for self-labels.
+            let records: Vec<(String, String, serde_json::Value)> = sqlx::query_as(
+                "SELECT uri, did, record FROM records WHERE uri = ANY($1)",
+            )
+            .bind(&uri_list)
+            .fetch_all(&state.db)
+            .await
+            .map_err(|e| mlua::Error::runtime(format!("record batch query failed: {e}")))?;
+
+            // Build result table keyed by URI.
+            let result = lua.create_table()?;
+
+            // Initialize empty arrays for each URI.
+            let mut counters: std::collections::HashMap<String, i32> = std::collections::HashMap::new();
+            for uri in &uri_list {
+                result.set(uri.as_str(), lua.create_table()?)?;
+                counters.insert(uri.clone(), 1);
+            }
+
+            // Add external labels.
+            for (src, uri, val, cts) in &rows {
+                let label = lua.create_table()?;
+                label.set("src", src.as_str())?;
+                label.set("uri", uri.as_str())?;
+                label.set("val", val.as_str())?;
+                label.set("cts", cts.to_rfc3339())?;
+
+                let uri_table: mlua::Table = result.get(uri.as_str())?;
+                let idx = counters.get(uri).copied().unwrap_or(1);
+                uri_table.set(idx, label)?;
+                counters.insert(uri.clone(), idx + 1);
+            }
+
+            // Add self-labels from records.
+            for (uri, did, record) in &records {
+                if let Some(labels) = record.get("labels")
+                    && let Some(values) = labels.get("values")
+                    && let Some(arr) = values.as_array()
+                {
+                    for item in arr {
+                        if let Some(val) = item.get("val").and_then(|v| v.as_str()) {
+                            let label = lua.create_table()?;
+                            label.set("src", did.as_str())?;
+                            label.set("uri", uri.as_str())?;
+                            label.set("val", val)?;
+                            label.set("cts", "")?;
+
+                            let uri_table: mlua::Table = result.get(uri.as_str())?;
+                            let idx = counters.get(uri).copied().unwrap_or(1);
+                            uri_table.set(idx, label)?;
+                            counters.insert(uri.clone(), idx + 1);
+                        }
+                    }
+                }
+            }
+
+            Ok(mlua::Value::Table(result))
+        }
+    })?;
+    atproto_table.set("get_labels_batch", get_labels_batch_fn)?;
+
     lua.globals().set("atproto", atproto_table)?;
     Ok(())
 }
@@ -49,12 +191,14 @@ mod tests {
             event_log_retention_days: 30,
         };
         let (tx, _) = watch::channel(vec![]);
+        let (labeler_tx, _) = watch::channel(());
         AppState {
             config,
             http: reqwest::Client::new(),
             db: sqlx::PgPool::connect_lazy("postgres://localhost/fake").unwrap(),
             lexicons: LexiconRegistry::new(),
             collections_tx: tx,
+            labeler_subscriptions_tx: labeler_tx,
         }
     }
 
diff --git a/src/lua/db_api.rs b/src/lua/db_api.rs
index eb47fc2..d8cf1ba 100644
--- a/src/lua/db_api.rs
+++ b/src/lua/db_api.rs
@@ -584,12 +584,14 @@ mod tests {
             event_log_retention_days: 30,
         };
         let (tx, _) = watch::channel(vec![]);
+        let (labeler_tx, _) = watch::channel(());
         AppState {
             config,
             http: reqwest::Client::new(),
             db: sqlx::PgPool::connect_lazy("postgres://localhost/fake").unwrap(),
             lexicons: LexiconRegistry::new(),
             collections_tx: tx,
+            labeler_subscriptions_tx: labeler_tx,
         }
     }
 
diff --git a/src/lua/execute.rs b/src/lua/execute.rs
index 411c114..a25effe 100644
--- a/src/lua/execute.rs
+++ b/src/lua/execute.rs
@@ -919,12 +919,14 @@ mod tests {
             event_log_retention_days: 30,
         };
         let (tx, _) = watch::channel(vec![]);
+        let (labeler_tx, _) = watch::channel(());
         AppState {
             config,
             http: reqwest::Client::new(),
             db: sqlx::PgPool::connect_lazy("postgres://localhost/fake").unwrap(),
             lexicons: LexiconRegistry::new(),
             collections_tx: tx,
+            labeler_subscriptions_tx: labeler_tx,
         }
     }
 
diff --git a/src/lua/http_api.rs b/src/lua/http_api.rs
index 0019a0d..306c92f 100644
--- a/src/lua/http_api.rs
+++ b/src/lua/http_api.rs
@@ -101,12 +101,14 @@ mod tests {
             event_log_retention_days: 30,
         };
         let (tx, _) = watch::channel(vec![]);
+        let (labeler_tx, _) = watch::channel(());
         AppState {
             config,
             http: reqwest::Client::new(),
             db: sqlx::PgPool::connect_lazy("postgres://localhost/fake").unwrap(),
             lexicons: LexiconRegistry::new(),
             collections_tx: tx,
+            labeler_subscriptions_tx: labeler_tx,
         }
     }
 
diff --git a/src/main.rs b/src/main.rs
index 6ac60c9..9540506 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -1,7 +1,7 @@
 use happyview::config::Config;
 use happyview::lexicon::{LexiconRegistry, ParsedLexicon, ProcedureAction};
 use happyview::resolve::{fetch_lexicon_from_pds, resolve_nsid_authority};
-use happyview::{AppState, server, tap};
+use happyview::{AppState, labeler, server, tap};
 use tokio::sync::watch;
 use tracing::{info, warn};
 
@@ -157,6 +157,7 @@ async fn main() {
     let initial_collections = lexicons.get_record_collections().await;
     let initial_collections_for_sync = initial_collections.clone();
     let (collections_tx, collections_rx) = watch::channel(initial_collections);
+    let (labeler_subscriptions_tx, labeler_subscriptions_rx) = watch::channel(());
 
     let state = AppState {
         config: config.clone(),
@@ -164,6 +165,7 @@ async fn main() {
         db,
         lexicons,
         collections_tx,
+        labeler_subscriptions_tx,
     };
 
     // Sync initial collections to Tap on startup.
@@ -186,6 +188,9 @@ async fn main() {
 
     tap::spawn(state.clone(), collections_rx);
 
+    labeler::spawn(state.clone(), labeler_subscriptions_rx);
+    tokio::spawn(labeler::spawn_label_gc(state.db.clone()));
+
     tokio::spawn(happyview::event_log::spawn_retention_cleanup(
         state.db.clone(),
         state.config.event_log_retention_days,
diff --git a/src/tap.rs b/src/tap.rs
index 5831ca2..44beb10 100644
--- a/src/tap.rs
+++ b/src/tap.rs
@@ -559,6 +559,8 @@ async fn handle_record_event(state: &AppState, record: &TapRecordEvent) {
                         },
                     )
                     .await;
+
+                    crate::labeler::backfill_labels_for_uri(Arc::new(state.clone()), uri.clone());
                 }
                 Err(e) => {
                     tracing::warn!(uri = %uri, "failed to upsert record: {e}");
diff --git a/tests/common/app.rs b/tests/common/app.rs
index c372272..3178811 100644
--- a/tests/common/app.rs
+++ b/tests/common/app.rs
@@ -59,6 +59,7 @@ impl TestApp {
 
         let initial_collections = lexicons.get_record_collections().await;
         let (collections_tx, _collections_rx) = watch::channel(initial_collections);
+        let (labeler_subscriptions_tx, _) = watch::channel(());
 
         let state = AppState {
             config,
@@ -66,6 +67,7 @@ impl TestApp {
             db: pool,
             lexicons,
             collections_tx,
+            labeler_subscriptions_tx,
         };
 
         let router = server::router(state.clone());
diff --git a/tests/common/db.rs b/tests/common/db.rs
index fec9f09..b7e3ae7 100644
--- a/tests/common/db.rs
+++ b/tests/common/db.rs
@@ -20,7 +20,7 @@ pub async fn test_pool() -> PgPool {
 /// Truncate all application tables, preserving schema.
 pub async fn truncate_all(pool: &PgPool) {
     sqlx::query(
-        "TRUNCATE records, lexicons, backfill_jobs, users, user_permissions, api_keys, event_logs, script_variables, dead_letter_hooks, record_refs RESTART IDENTITY CASCADE",
+        "TRUNCATE records, lexicons, backfill_jobs, users, user_permissions, api_keys, event_logs, script_variables, dead_letter_hooks, record_refs, labeler_subscriptions, labels RESTART IDENTITY CASCADE",
     )
     .execute(pool)
     .await
diff --git a/tests/e2e_labelers.rs b/tests/e2e_labelers.rs
new file mode 100644
index 0000000..881942f
--- /dev/null
+++ b/tests/e2e_labelers.rs
@@ -0,0 +1,387 @@
+mod common;
+
+use axum::body::Body;
+use axum::http::{Request, StatusCode};
+use http_body_util::BodyExt;
+use serde_json::{Value, json};
+use serial_test::serial;
+use tower::ServiceExt;
+
+use common::app::TestApp;
+use common::auth::admin_auth_header;
+
+// ---------------------------------------------------------------------------
+// Helpers
+// ---------------------------------------------------------------------------
+
+async fn json_body(resp: axum::response::Response) -> Value {
+    let body = resp.into_body().collect().await.unwrap().to_bytes();
+    serde_json::from_slice(&body).unwrap()
+}
+
+fn admin_get(uri: &str, token: &str) -> Request<Body> {
+    let (hname, hval) = admin_auth_header(token);
+    Request::builder()
+        .uri(uri)
+        .header(hname, hval)
+        .body(Body::empty())
+        .unwrap()
+}
+
+fn admin_post(uri: &str, token: &str, body: &Value) -> Request<Body> {
+    let (hname, hval) = admin_auth_header(token);
+    Request::builder()
+        .method("POST")
+        .uri(uri)
+        .header(hname, hval)
+        .header("content-type", "application/json")
+        .body(Body::from(serde_json::to_vec(body).unwrap()))
+        .unwrap()
+}
+
+fn admin_patch(uri: &str, token: &str, body: &Value) -> Request<Body> {
+    let (hname, hval) = admin_auth_header(token);
+    Request::builder()
+        .method("PATCH")
+        .uri(uri)
+        .header(hname, hval)
+        .header("content-type", "application/json")
+        .body(Body::from(serde_json::to_vec(body).unwrap()))
+        .unwrap()
+}
+
+fn admin_delete(uri: &str, token: &str) -> Request<Body> {
+    let (hname, hval) = admin_auth_header(token);
+    Request::builder()
+        .method("DELETE")
+        .uri(uri)
+        .header(hname, hval)
+        .body(Body::empty())
+        .unwrap()
+}
+
+// ---------------------------------------------------------------------------
+// POST /admin/labelers
+// ---------------------------------------------------------------------------
+
+#[tokio::test]
+#[serial]
+async fn labeler_add_returns_201() {
+    let app = TestApp::new().await;
+    app.mock_admin_userinfo().await;
+
+    let body = json!({ "did": "did:plc:labeler1" });
+
+    let resp = app
+        .router
+        .oneshot(admin_post("/admin/labelers", &app.admin_token, &body))
+        .await
+        .unwrap();
+
+    assert_eq!(resp.status(), StatusCode::CREATED);
+}
+
+#[tokio::test]
+#[serial]
+async fn labeler_add_upsert_reactivates() {
+    let app = TestApp::new().await;
+    app.mock_admin_userinfo().await;
+
+    let body = json!({ "did": "did:plc:labeler1" });
+
+    // First add
+    app.router
+        .clone()
+        .oneshot(admin_post("/admin/labelers", &app.admin_token, &body))
+        .await
+        .unwrap();
+
+    // Pause it
+    app.router
+        .clone()
+        .oneshot(admin_patch(
+            "/admin/labelers/did:plc:labeler1",
+            &app.admin_token,
+            &json!({ "status": "paused" }),
+        ))
+        .await
+        .unwrap();
+
+    // Re-add (upsert should reactivate)
+    let resp = app
+        .router
+        .clone()
+        .oneshot(admin_post("/admin/labelers", &app.admin_token, &body))
+        .await
+        .unwrap();
+
+    assert_eq!(resp.status(), StatusCode::CREATED);
+
+    // Verify it's active again
+    let resp = app
+        .router
+        .oneshot(admin_get("/admin/labelers", &app.admin_token))
+        .await
+        .unwrap();
+
+    let json = json_body(resp).await;
+    let labelers = json.as_array().unwrap();
+    assert_eq!(labelers.len(), 1);
+    assert_eq!(labelers[0]["status"], "active");
+}
+
+// ---------------------------------------------------------------------------
+// GET /admin/labelers
+// ---------------------------------------------------------------------------
+
+#[tokio::test]
+#[serial]
+async fn labeler_list_empty() {
+    let app = TestApp::new().await;
+    app.mock_admin_userinfo().await;
+
+    let resp = app
+        .router
+        .oneshot(admin_get("/admin/labelers", &app.admin_token))
+        .await
+        .unwrap();
+
+    assert_eq!(resp.status(), StatusCode::OK);
+    let json = json_body(resp).await;
+    assert!(json.as_array().unwrap().is_empty());
+}
+
+#[tokio::test]
+#[serial]
+async fn labeler_list_returns_added() {
+    let app = TestApp::new().await;
+    app.mock_admin_userinfo().await;
+
+    app.router
+        .clone()
+        .oneshot(admin_post(
+            "/admin/labelers",
+            &app.admin_token,
+            &json!({ "did": "did:plc:lab1" }),
+        ))
+        .await
+        .unwrap();
+
+    app.router
+        .clone()
+        .oneshot(admin_post(
+            "/admin/labelers",
+            &app.admin_token,
+            &json!({ "did": "did:plc:lab2" }),
+        ))
+        .await
+        .unwrap();
+
+    let resp = app
+        .router
+        .oneshot(admin_get("/admin/labelers", &app.admin_token))
+        .await
+        .unwrap();
+
+    assert_eq!(resp.status(), StatusCode::OK);
+    let json = json_body(resp).await;
+    let labelers = json.as_array().unwrap();
+    assert_eq!(labelers.len(), 2);
+    assert_eq!(labelers[0]["did"], "did:plc:lab1");
+    assert_eq!(labelers[0]["status"], "active");
+    assert_eq!(labelers[1]["did"], "did:plc:lab2");
+}
+
+// ---------------------------------------------------------------------------
+// PATCH /admin/labelers/{did}
+// ---------------------------------------------------------------------------
+
+#[tokio::test]
+#[serial]
+async fn labeler_update_status() {
+    let app = TestApp::new().await;
+    app.mock_admin_userinfo().await;
+
+    app.router
+        .clone()
+        .oneshot(admin_post(
+            "/admin/labelers",
+            &app.admin_token,
+            &json!({ "did": "did:plc:lab1" }),
+        ))
+        .await
+        .unwrap();
+
+    let resp = app
+        .router
+        .clone()
+        .oneshot(admin_patch(
+            "/admin/labelers/did:plc:lab1",
+            &app.admin_token,
+            &json!({ "status": "paused" }),
+        ))
+        .await
+        .unwrap();
+
+    assert_eq!(resp.status(), StatusCode::NO_CONTENT);
+
+    // Verify status changed
+    let resp = app
+        .router
+        .oneshot(admin_get("/admin/labelers", &app.admin_token))
+        .await
+        .unwrap();
+
+    let json = json_body(resp).await;
+    let labelers = json.as_array().unwrap();
+    assert_eq!(labelers[0]["status"], "paused");
+}
+
+#[tokio::test]
+#[serial]
+async fn labeler_update_not_found() {
+    let app = TestApp::new().await;
+    app.mock_admin_userinfo().await;
+
+    let resp = app
+        .router
+        .oneshot(admin_patch(
+            "/admin/labelers/did:plc:nonexistent",
+            &app.admin_token,
+            &json!({ "status": "paused" }),
+        ))
+        .await
+        .unwrap();
+
+    assert_eq!(resp.status(), StatusCode::NOT_FOUND);
+}
+
+// ---------------------------------------------------------------------------
+// DELETE /admin/labelers/{did}
+// ---------------------------------------------------------------------------
+
+#[tokio::test]
+#[serial]
+async fn labeler_delete_returns_204() {
+    let app = TestApp::new().await;
+    app.mock_admin_userinfo().await;
+
+    app.router
+        .clone()
+        .oneshot(admin_post(
+            "/admin/labelers",
+            &app.admin_token,
+            &json!({ "did": "did:plc:lab1" }),
+        ))
+        .await
+        .unwrap();
+
+    let resp = app
+        .router
+        .clone()
+        .oneshot(admin_delete(
+            "/admin/labelers/did:plc:lab1",
+            &app.admin_token,
+        ))
+        .await
+        .unwrap();
+
+    assert_eq!(resp.status(), StatusCode::NO_CONTENT);
+
+    // Verify it's gone
+    let resp = app
+        .router
+        .oneshot(admin_get("/admin/labelers", &app.admin_token))
+        .await
+        .unwrap();
+
+    let json = json_body(resp).await;
+    assert!(json.as_array().unwrap().is_empty());
+}
+
+#[tokio::test]
+#[serial]
+async fn labeler_delete_not_found() {
+    let app = TestApp::new().await;
+    app.mock_admin_userinfo().await;
+
+    let resp = app
+        .router
+        .oneshot(admin_delete(
+            "/admin/labelers/did:plc:nonexistent",
+            &app.admin_token,
+        ))
+        .await
+        .unwrap();
+
+    assert_eq!(resp.status(), StatusCode::NOT_FOUND);
+}
+
+#[tokio::test]
+#[serial]
+async fn labeler_delete_removes_labels() {
+    let app = TestApp::new().await;
+    app.mock_admin_userinfo().await;
+
+    // Add a labeler
+    app.router
+        .clone()
+        .oneshot(admin_post(
+            "/admin/labelers",
+            &app.admin_token,
+            &json!({ "did": "did:plc:lab1" }),
+        ))
+        .await
+        .unwrap();
+
+    // Seed some labels from that labeler
+    sqlx::query("INSERT INTO labels (src, uri, val, cts) VALUES ($1, $2, $3, NOW())")
+        .bind("did:plc:lab1")
+        .bind("at://did:plc:user/test.collection/rkey1")
+        .bind("adult-content")
+        .execute(&app.state.db)
+        .await
+        .unwrap();
+
+    // Delete the labeler
+    app.router
+        .clone()
+        .oneshot(admin_delete(
+            "/admin/labelers/did:plc:lab1",
+            &app.admin_token,
+        ))
+        .await
+        .unwrap();
+
+    // Verify labels were also removed
+    let count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM labels WHERE src = $1")
+        .bind("did:plc:lab1")
+        .fetch_one(&app.state.db)
+        .await
+        .unwrap();
+
+    assert_eq!(count.0, 0);
+}
+
+// ---------------------------------------------------------------------------
+// Auth
+// ---------------------------------------------------------------------------
+
+#[tokio::test]
+#[serial]
+async fn labeler_no_auth_returns_401() {
+    let app = TestApp::new().await;
+
+    let resp = app
+        .router
+        .oneshot(
+            Request::builder()
+                .uri("/admin/labelers")
+                .body(Body::empty())
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+
+    assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
+}
diff --git a/tests/lua_atproto_api.rs b/tests/lua_atproto_api.rs
new file mode 100644
index 0000000..eea228a
--- /dev/null
+++ b/tests/lua_atproto_api.rs
@@ -0,0 +1,386 @@
+mod common;
+
+use happyview::AppState;
+use happyview::config::Config;
+use happyview::lexicon::LexiconRegistry;
+use serial_test::serial;
+use tokio::sync::watch;
+
+use common::db;
+
+/// Build an AppState backed by a real Postgres pool.
+async fn test_state_with_pool(pool: sqlx::PgPool) -> AppState {
+    let config = Config {
+        host: "127.0.0.1".into(),
+        port: 3000,
+        database_url: String::new(),
+        aip_url: String::new(),
+        aip_public_url: String::new(),
+        tap_url: String::new(),
+        tap_admin_password: None,
+        relay_url: String::new(),
+        plc_url: String::new(),
+        static_dir: String::new(),
+        event_log_retention_days: 30,
+    };
+    let (tx, _) = watch::channel(vec![]);
+    let (labeler_tx, _) = watch::channel(());
+    AppState {
+        config,
+        http: reqwest::Client::new(),
+        db: pool,
+        lexicons: LexiconRegistry::new(),
+        collections_tx: tx,
+        labeler_subscriptions_tx: labeler_tx,
+    }
+}
+
+/// Seed a record into the records table.
+async fn seed_record(pool: &sqlx::PgPool, uri: &str, did: &str, record: serde_json::Value) {
+    sqlx::query(
+        "INSERT INTO records (uri, did, collection, rkey, record, cid) VALUES ($1, $2, $3, $4, $5, $6)",
+    )
+    .bind(uri)
+    .bind(did)
+    .bind("test.collection")
+    .bind("rkey1")
+    .bind(record)
+    .bind("bafytest")
+    .execute(pool)
+    .await
+    .expect("failed to seed record");
+}
+
+/// Seed a label into the labels table.
+async fn seed_label(pool: &sqlx::PgPool, src: &str, uri: &str, val: &str, exp: Option<&str>) {
+    if let Some(exp) = exp {
+        sqlx::query(
+            "INSERT INTO labels (src, uri, val, cts, exp) VALUES ($1, $2, $3, NOW(), $4::timestamptz)",
+        )
+        .bind(src)
+        .bind(uri)
+        .bind(val)
+        .bind(exp)
+        .execute(pool)
+        .await
+        .expect("failed to seed label");
+    } else {
+        sqlx::query("INSERT INTO labels (src, uri, val, cts) VALUES ($1, $2, $3, NOW())")
+            .bind(src)
+            .bind(uri)
+            .bind(val)
+            .execute(pool)
+            .await
+            .expect("failed to seed label");
+    }
+}
+
+// ---------------------------------------------------------------------------
+// get_labels tests
+// ---------------------------------------------------------------------------
+
+#[tokio::test]
+#[serial]
+async fn get_labels_returns_external_labels() {
+    let pool = db::test_pool().await;
+    db::truncate_all(&pool).await;
+
+    let uri = "at://did:plc:test/test.collection/rkey1";
+    seed_record(
+        &pool,
+        uri,
+        "did:plc:test",
+        serde_json::json!({"name": "test"}),
+    )
+    .await;
+    seed_label(&pool, "did:plc:labeler1", uri, "adult-content", None).await;
+    seed_label(&pool, "did:plc:labeler1", uri, "violence", None).await;
+
+    let state = test_state_with_pool(pool).await;
+
+    // atproto_api is pub(crate) so we test the underlying queries directly.
+    let rows: Vec<(String, String, String)> = sqlx::query_as(
+        "SELECT src, uri, val FROM labels WHERE uri = $1 AND (exp IS NULL OR exp > NOW())",
+    )
+    .bind(uri)
+    .fetch_all(&state.db)
+    .await
+    .unwrap();
+
+    assert_eq!(rows.len(), 2);
+    assert_eq!(rows[0].2, "adult-content");
+    assert_eq!(rows[1].2, "violence");
+}
+
+#[tokio::test]
+#[serial]
+async fn get_labels_filters_expired() {
+    let pool = db::test_pool().await;
+    db::truncate_all(&pool).await;
+
+    let uri = "at://did:plc:test/test.collection/rkey1";
+    seed_record(
+        &pool,
+        uri,
+        "did:plc:test",
+        serde_json::json!({"name": "test"}),
+    )
+    .await;
+
+    // Active label
+    seed_label(&pool, "did:plc:labeler1", uri, "nudity", None).await;
+    // Expired label (past date)
+    seed_label(
+        &pool,
+        "did:plc:labeler1",
+        uri,
+        "spam",
+        Some("2020-01-01T00:00:00Z"),
+    )
+    .await;
+
+    let state = test_state_with_pool(pool).await;
+
+    // Query with expiry filter (same as get_labels does internally)
+    let rows: Vec<(String, String, String)> = sqlx::query_as(
+        "SELECT src, uri, val FROM labels WHERE uri = $1 AND (exp IS NULL OR exp > NOW())",
+    )
+    .bind(uri)
+    .fetch_all(&state.db)
+    .await
+    .unwrap();
+
+    assert_eq!(rows.len(), 1);
+    assert_eq!(rows[0].2, "nudity");
+}
+
+#[tokio::test]
+#[serial]
+async fn get_labels_includes_self_labels() {
+    let pool = db::test_pool().await;
+    db::truncate_all(&pool).await;
+
+    let uri = "at://did:plc:author/test.collection/rkey1";
+    let record = serde_json::json!({
+        "name": "test",
+        "labels": {
+            "values": [
+                { "val": "sexual" },
+                { "val": "graphic-media" }
+            ]
+        }
+    });
+    seed_record(&pool, uri, "did:plc:author", record.clone()).await;
+
+    // Verify self-labels can be extracted from record JSON
+    let fetched: Option<(String, serde_json::Value)> =
+        sqlx::query_as("SELECT did, record FROM records WHERE uri = $1")
+            .bind(uri)
+            .fetch_optional(&pool)
+            .await
+            .unwrap();
+
+    let (did, rec) = fetched.unwrap();
+    assert_eq!(did, "did:plc:author");
+
+    let self_labels: Vec<&str> = rec
+        .get("labels")
+        .and_then(|l| l.get("values"))
+        .and_then(|v| v.as_array())
+        .unwrap()
+        .iter()
+        .filter_map(|item| item.get("val").and_then(|v| v.as_str()))
+        .collect();
+
+    assert_eq!(self_labels, vec!["sexual", "graphic-media"]);
+}
+
+#[tokio::test]
+#[serial]
+async fn get_labels_empty_for_unlabeled_record() {
+    let pool = db::test_pool().await;
+    db::truncate_all(&pool).await;
+
+    let uri = "at://did:plc:test/test.collection/rkey1";
+    seed_record(
+        &pool,
+        uri,
+        "did:plc:test",
+        serde_json::json!({"name": "test"}),
+    )
+    .await;
+
+    let rows: Vec<(String, String, String)> = sqlx::query_as(
+        "SELECT src, uri, val FROM labels WHERE uri = $1 AND (exp IS NULL OR exp > NOW())",
+    )
+    .bind(uri)
+    .fetch_all(&pool)
+    .await
+    .unwrap();
+
+    assert!(rows.is_empty());
+}
+
+// ---------------------------------------------------------------------------
+// get_labels_batch tests
+// ---------------------------------------------------------------------------
+
+#[tokio::test]
+#[serial]
+async fn get_labels_batch_returns_labels_per_uri() {
+    let pool = db::test_pool().await;
+    db::truncate_all(&pool).await;
+
+    let uri1 = "at://did:plc:test/test.collection/rkey1";
+    let uri2 = "at://did:plc:test/test.collection/rkey2";
+
+    seed_record(
+        &pool,
+        uri1,
+        "did:plc:test",
+        serde_json::json!({"name": "one"}),
+    )
+    .await;
+
+    // Use different rkey for second record to avoid PK conflict
+    sqlx::query(
+        "INSERT INTO records (uri, did, collection, rkey, record, cid) VALUES ($1, $2, $3, $4, $5, $6)",
+    )
+    .bind(uri2)
+    .bind("did:plc:test")
+    .bind("test.collection")
+    .bind("rkey2")
+    .bind(serde_json::json!({"name": "two"}))
+    .bind("bafytest2")
+    .execute(&pool)
+    .await
+    .unwrap();
+
+    seed_label(&pool, "did:plc:labeler1", uri1, "nudity", None).await;
+    seed_label(&pool, "did:plc:labeler1", uri2, "spam", None).await;
+    seed_label(&pool, "did:plc:labeler2", uri2, "violence", None).await;
+
+    let uris = vec![uri1.to_string(), uri2.to_string()];
+
+    // Batch query (same as get_labels_batch does internally)
+    let rows: Vec<(String, String, String)> = sqlx::query_as(
+        "SELECT src, uri, val FROM labels WHERE uri = ANY($1) AND (exp IS NULL OR exp > NOW())",
+    )
+    .bind(&uris)
+    .fetch_all(&pool)
+    .await
+    .unwrap();
+
+    // uri1 has 1 label, uri2 has 2 labels
+    let uri1_labels: Vec<_> = rows.iter().filter(|r| r.1 == uri1).collect();
+    let uri2_labels: Vec<_> = rows.iter().filter(|r| r.1 == uri2).collect();
+
+    assert_eq!(uri1_labels.len(), 1);
+    assert_eq!(uri1_labels[0].2, "nudity");
+    assert_eq!(uri2_labels.len(), 2);
+}
+
+#[tokio::test]
+#[serial]
+async fn get_labels_batch_empty_for_no_labels() {
+    let pool = db::test_pool().await;
+    db::truncate_all(&pool).await;
+
+    let uris = vec![
+        "at://did:plc:test/test.collection/rkey1".to_string(),
+        "at://did:plc:test/test.collection/rkey2".to_string(),
+    ];
+
+    let rows: Vec<(String, String, String)> = sqlx::query_as(
+        "SELECT src, uri, val FROM labels WHERE uri = ANY($1) AND (exp IS NULL OR exp > NOW())",
+    )
+    .bind(&uris)
+    .fetch_all(&pool)
+    .await
+    .unwrap();
+
+    assert!(rows.is_empty());
+}
+
+// ---------------------------------------------------------------------------
+// Label negation (materialized state)
+// ---------------------------------------------------------------------------
+
+#[tokio::test]
+#[serial]
+async fn label_negation_removes_row() {
+    let pool = db::test_pool().await;
+    db::truncate_all(&pool).await;
+
+    let uri = "at://did:plc:test/test.collection/rkey1";
+
+    // Add a label
+    seed_label(&pool, "did:plc:labeler1", uri, "nudity", None).await;
+
+    // Verify it exists
+    let count: (i64,) =
+        sqlx::query_as("SELECT COUNT(*) FROM labels WHERE src = $1 AND uri = $2 AND val = $3")
+            .bind("did:plc:labeler1")
+            .bind(uri)
+            .bind("nudity")
+            .fetch_one(&pool)
+            .await
+            .unwrap();
+    assert_eq!(count.0, 1);
+
+    // Simulate negation (same logic as labeler.rs)
+    sqlx::query("DELETE FROM labels WHERE src = $1 AND uri = $2 AND val = $3")
+        .bind("did:plc:labeler1")
+        .bind(uri)
+        .bind("nudity")
+        .execute(&pool)
+        .await
+        .unwrap();
+
+    // Verify it's gone
+    let count: (i64,) =
+        sqlx::query_as("SELECT COUNT(*) FROM labels WHERE src = $1 AND uri = $2 AND val = $3")
+            .bind("did:plc:labeler1")
+            .bind(uri)
+            .bind("nudity")
+            .fetch_one(&pool)
+            .await
+            .unwrap();
+    assert_eq!(count.0, 0);
+}
+
+// ---------------------------------------------------------------------------
+// Label upsert (idempotent)
+// ---------------------------------------------------------------------------
+
+#[tokio::test]
+#[serial]
+async fn label_upsert_is_idempotent() {
+    let pool = db::test_pool().await;
+    db::truncate_all(&pool).await;
+
+    let uri = "at://did:plc:test/test.collection/rkey1";
+
+    // Insert same label twice (upsert pattern from labeler.rs)
+    for _ in 0..2 {
+        sqlx::query(
+            "INSERT INTO labels (src, uri, val, cts) VALUES ($1, $2, $3, NOW()) ON CONFLICT (src, uri, val) DO UPDATE SET cts = EXCLUDED.cts",
+        )
+        .bind("did:plc:labeler1")
+        .bind(uri)
+        .bind("nudity")
+        .execute(&pool)
+        .await
+        .unwrap();
+    }
+
+    let count: (i64,) =
+        sqlx::query_as("SELECT COUNT(*) FROM labels WHERE src = $1 AND uri = $2 AND val = $3")
+            .bind("did:plc:labeler1")
+            .bind(uri)
+            .bind("nudity")
+            .fetch_one(&pool)
+            .await
+            .unwrap();
+    assert_eq!(count.0, 1);
+}
diff --git a/tests/lua_db_api.rs b/tests/lua_db_api.rs
index d23de71..0da3294 100644
--- a/tests/lua_db_api.rs
+++ b/tests/lua_db_api.rs
@@ -27,12 +27,14 @@ async fn test_state_with_pool(pool: sqlx::PgPool) -> AppState {
         event_log_retention_days: 30,
     };
     let (tx, _) = watch::channel(vec![]);
+    let (labeler_tx, _) = watch::channel(());
     AppState {
         config,
         http: reqwest::Client::new(),
         db: pool,
         lexicons: LexiconRegistry::new(),
         collections_tx: tx,
+        labeler_subscriptions_tx: labeler_tx,
     }
 }
 
-- 
2.51.2