Canadian cities labeler over Bluesky DMs

Users DM @cities.pds.cosocial.ca their city and get a signed account label.
Serves com.atproto.label.queryLabels and subscribeLabels, built on jacquard.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

.env.example

@@ -0,0 +1,9 @@
+LABELER_HANDLE=cities.pds.cosocial.ca
+# Settings → Privacy and security → App passwords, with "Allow access to your direct messages"
+LABELER_APP_PASSWORD=
+# from `cargo run --bin keygen`
+SIGNING_KEY_HEX=
+# PDS_URL=https://pds.cosocial.ca
+# PORT=4100
+# DB_PATH=labeler.sqlite
+# CITIES_PATH=data/cities.json

.gitignore

@@ -0,0 +1,4 @@
+/target
+.env
+*.sqlite
+*.sqlite-*

Cargo.toml

@@ -0,0 +1,22 @@
+[package]
+name = "cities-labeler"
+version = "0.1.0"
+edition = "2024"
+publish = false
+default-run = "cities-labeler"
+
+[dependencies]
+anyhow = "1"
+axum = { version = "0.8", features = ["ws"] }
+dotenvy = "0.15"
+hex = "0.4"
+jacquard = { version = "=0.12.2", default-features = false, features = ["api_bluesky", "streaming"] }
+jacquard-axum = "=0.12.2"
+k256 = { version = "0.13", features = ["ecdsa"] }
+rand_core = { version = "0.6", features = ["getrandom"] }
+rusqlite = { version = "0.37", features = ["bundled"] }
+serde = { version = "1", features = ["derive"] }
+serde_ipld_dagcbor = "0.6"
+serde_json = "1"
+tokio = { version = "1", features = ["full"] }
+unicode-normalization = "0.1"

README.md

@@ -0,0 +1,41 @@
+# cosocial cities labeler
+
+An atproto labeler that puts a Canadian city badge on accounts that ask for one over Bluesky DMs.
+
+1. Subscribe to `@cities.pds.cosocial.ca` (the Subscribe button on its profile).
+2. Follow it.
+3. DM it your city, e.g. `Halifax` or `Richmond, BC`. Send `remove` to clear it.
+
+The labeler polls its DMs, matches the city against `data/cities.json`, and signs an account
+label (e.g. `halifax-ns`), negating any previous one. Labels are stored in SQLite and served over
+`com.atproto.label.subscribeLabels` and `queryLabels`.
+
+```
+src/
+  dms.rs       DM polling, onboarding, replies
+  cities.rs    city list and matching
+  labeler.rs   set_city: negate the old label, apply the new one
+  labels.rs    label signing
+  server.rs    queryLabels + subscribeLabels (axum, jacquard-axum)
+  db.rs        SQLite
+  bin/keygen.rs
+```
+
+Built on [jacquard](https://tangled.org/nonbinary.computer/jacquard) for atproto types and XRPC.
+
+## Local test
+
+```sh
+cp .env.example .env      # set LABELER_APP_PASSWORD
+cargo run --bin keygen    # paste SIGNING_KEY_HEX into .env
+cargo run                 # :4100
+```
+
+`rust-toolchain.toml` pins the Rust version; rustup installs it on first build.
+
+The labeler account must allow DMs from everyone (Settings → Privacy and security → Chat).
+DM it from a test account, then check
+`curl 'localhost:4100/xrpc/com.atproto.label.queryLabels?uriPatterns=<did>'`.
+
+Labels won't appear in the Bluesky app until the labeler is deployed publicly and its DID
+document has the `#atproto_label` key and `#atproto_labeler` service.

data/cities.json

@@ -0,0 +1,54 @@
+[
+  { "id": "toronto-on", "name": "Toronto", "province": "ON" },
+  { "id": "montreal-qc", "name": "Montréal", "province": "QC" },
+  { "id": "calgary-ab", "name": "Calgary", "province": "AB" },
+  { "id": "ottawa-on", "name": "Ottawa", "province": "ON" },
+  { "id": "edmonton-ab", "name": "Edmonton", "province": "AB" },
+  { "id": "winnipeg-mb", "name": "Winnipeg", "province": "MB" },
+  { "id": "mississauga-on", "name": "Mississauga", "province": "ON" },
+  { "id": "vancouver-bc", "name": "Vancouver", "province": "BC" },
+  { "id": "brampton-on", "name": "Brampton", "province": "ON" },
+  { "id": "hamilton-on", "name": "Hamilton", "province": "ON" },
+  { "id": "surrey-bc", "name": "Surrey", "province": "BC" },
+  { "id": "quebec-qc", "name": "Québec", "province": "QC" },
+  { "id": "halifax-ns", "name": "Halifax", "province": "NS" },
+  { "id": "laval-qc", "name": "Laval", "province": "QC" },
+  { "id": "london-on", "name": "London", "province": "ON" },
+  { "id": "markham-on", "name": "Markham", "province": "ON" },
+  { "id": "vaughan-on", "name": "Vaughan", "province": "ON" },
+  { "id": "gatineau-qc", "name": "Gatineau", "province": "QC" },
+  { "id": "saskatoon-sk", "name": "Saskatoon", "province": "SK" },
+  { "id": "kitchener-on", "name": "Kitchener", "province": "ON" },
+  { "id": "longueuil-qc", "name": "Longueuil", "province": "QC" },
+  { "id": "burnaby-bc", "name": "Burnaby", "province": "BC" },
+  { "id": "windsor-on", "name": "Windsor", "province": "ON" },
+  { "id": "regina-sk", "name": "Regina", "province": "SK" },
+  { "id": "oakville-on", "name": "Oakville", "province": "ON" },
+  { "id": "richmond-bc", "name": "Richmond", "province": "BC" },
+  { "id": "richmond-hill-on", "name": "Richmond Hill", "province": "ON" },
+  { "id": "burlington-on", "name": "Burlington", "province": "ON" },
+  { "id": "oshawa-on", "name": "Oshawa", "province": "ON" },
+  { "id": "sherbrooke-qc", "name": "Sherbrooke", "province": "QC" },
+  { "id": "greater-sudbury-on", "name": "Greater Sudbury", "province": "ON" },
+  { "id": "abbotsford-bc", "name": "Abbotsford", "province": "BC" },
+  { "id": "levis-qc", "name": "Lévis", "province": "QC" },
+  { "id": "coquitlam-bc", "name": "Coquitlam", "province": "BC" },
+  { "id": "barrie-on", "name": "Barrie", "province": "ON" },
+  { "id": "saguenay-qc", "name": "Saguenay", "province": "QC" },
+  { "id": "kelowna-bc", "name": "Kelowna", "province": "BC" },
+  { "id": "guelph-on", "name": "Guelph", "province": "ON" },
+  { "id": "trois-rivieres-qc", "name": "Trois-Rivières", "province": "QC" },
+  { "id": "whitby-on", "name": "Whitby", "province": "ON" },
+  { "id": "cambridge-on", "name": "Cambridge", "province": "ON" },
+  { "id": "st-catharines-on", "name": "St. Catharines", "province": "ON" },
+  { "id": "kingston-on", "name": "Kingston", "province": "ON" },
+  { "id": "victoria-bc", "name": "Victoria", "province": "BC" },
+  { "id": "moncton-nb", "name": "Moncton", "province": "NB" },
+  { "id": "saint-john-nb", "name": "Saint John", "province": "NB" },
+  { "id": "fredericton-nb", "name": "Fredericton", "province": "NB" },
+  { "id": "st-johns-nl", "name": "St. John's", "province": "NL" },
+  { "id": "charlottetown-pe", "name": "Charlottetown", "province": "PE" },
+  { "id": "whitehorse-yt", "name": "Whitehorse", "province": "YT" },
+  { "id": "yellowknife-nt", "name": "Yellowknife", "province": "NT" },
+  { "id": "iqaluit-nu", "name": "Iqaluit", "province": "NU" }
+]

rust-toolchain.toml

@@ -0,0 +1,4 @@
+[toolchain]
+channel = "1.91.0"
+profile = "minimal"
+components = ["rustfmt", "clippy", "rust-analyzer", "rust-src"]

rustfmt.toml

@@ -0,0 +1 @@
+edition = "2024"

src/bin/keygen.rs

@@ -0,0 +1,14 @@
+//! Generates the label signing key. The public multibase goes into the
+//! labeler's DID document as the `#atproto_label` verification method.
+use jacquard::common::types::crypto::multikey;
+use k256::ecdsa::SigningKey;
+use rand_core::OsRng;
+
+fn main() {
+    let key = SigningKey::random(&mut OsRng);
+    let public = key.verifying_key().to_encoded_point(true);
+    let multibase = multikey(0xE7, public.as_bytes());
+    println!("SIGNING_KEY_HEX={}", hex::encode(key.to_bytes()));
+    println!("# public key (did:key): did:key:{multibase}");
+    println!("# publicKeyMultibase:   {multibase}");
+}

src/cities.rs

@@ -0,0 +1,103 @@
+use std::collections::HashMap;
+
+use anyhow::{Result, bail};
+use serde::Deserialize;
+use unicode_normalization::{UnicodeNormalization, char::is_combining_mark};
+
+#[derive(Debug, Clone, Deserialize)]
+pub struct City {
+    pub id: String,
+    pub name: String,
+    pub province: String,
+}
+
+impl City {
+    pub fn display(&self) -> String {
+        format!("{}, {}", self.name, self.province)
+    }
+}
+
+pub struct Match<'a> {
+    pub exact: Vec<&'a City>,
+    pub suggestions: Vec<&'a City>,
+}
+
+struct Keyed {
+    city: City,
+    name: String,
+    keys: [String; 2],
+}
+
+pub struct Cities {
+    keyed: Vec<Keyed>,
+    by_id: HashMap<String, usize>,
+}
+
+/// "Trois-Rivières, QC" → "trois rivieres qc", "St. John's" → "st johns"
+fn fold(s: &str) -> String {
+    let stripped: String = s
+        .nfd()
+        .filter(|c| !is_combining_mark(*c) && !matches!(c, '\'' | '’' | '.'))
+        .flat_map(char::to_lowercase)
+        .map(|c| if c.is_ascii_alphanumeric() { c } else { ' ' })
+        .collect();
+    stripped.split_whitespace().collect::<Vec<_>>().join(" ")
+}
+
+/// Custom label identifiers must be lowercase letters and hyphens.
+fn is_label_id(id: &str) -> bool {
+    (1..=100).contains(&id.len()) && id.bytes().all(|b| b.is_ascii_lowercase() || b == b'-')
+}
+
+impl Cities {
+    pub fn load(path: &str) -> Result<Self> {
+        let cities: Vec<City> = serde_json::from_str(&std::fs::read_to_string(path)?)?;
+        let mut by_id = HashMap::new();
+        let mut keyed = Vec::with_capacity(cities.len());
+        for city in cities {
+            if !is_label_id(&city.id) {
+                bail!("invalid city id: {}", city.id);
+            }
+            if by_id.insert(city.id.clone(), keyed.len()).is_some() {
+                bail!("duplicate city id: {}", city.id);
+            }
+            let name = fold(&city.name);
+            let keys = [
+                name.clone(),
+                fold(&format!("{} {}", city.name, city.province)),
+            ];
+            keyed.push(Keyed { city, name, keys });
+        }
+        Ok(Self { keyed, by_id })
+    }
+
+    pub fn len(&self) -> usize {
+        self.keyed.len()
+    }
+
+    pub fn get(&self, id: &str) -> Option<&City> {
+        self.by_id.get(id).map(|&i| &self.keyed[i].city)
+    }
+
+    /// Exact matches by name or "name province"; otherwise name prefixes.
+    pub fn find(&self, text: &str) -> Match<'_> {
+        let q = fold(text);
+        let exact: Vec<_> = self
+            .keyed
+            .iter()
+            .filter(|k| k.keys.contains(&q))
+            .map(|k| &k.city)
+            .collect();
+        let suggestions = if exact.is_empty() && !q.is_empty() {
+            self.keyed
+                .iter()
+                .filter(|k| k.name.starts_with(&q))
+                .map(|k| &k.city)
+                .take(5)
+                .collect()
+        } else {
+            Vec::new()
+        };
+        Match { exact, suggestions }
+    }
+}

src/config.rs

@@ -0,0 +1,33 @@
+use anyhow::{Context, Result};
+
+pub struct Config {
+    pub handle: String,
+    /// App password with DM access enabled.
+    pub app_password: String,
+    pub signing_key_hex: String,
+    pub port: u16,
+    pub db_path: String,
+    pub cities_path: String,
+}
+
+fn required(name: &str) -> Result<String> {
+    std::env::var(name).with_context(|| format!("missing env var {name}"))
+}
+
+fn optional(name: &str, default: &str) -> String {
+    std::env::var(name).unwrap_or_else(|_| default.to_owned())
+}
+
+impl Config {
+    pub fn from_env() -> Result<Self> {
+        dotenvy::dotenv().ok();
+        Ok(Self {
+            handle: optional("LABELER_HANDLE", "cities.pds.cosocial.ca"),
+            app_password: required("LABELER_APP_PASSWORD")?,
+            signing_key_hex: required("SIGNING_KEY_HEX")?,
+            port: optional("PORT", "4100").parse().context("PORT")?,
+            db_path: optional("DB_PATH", "labeler.sqlite"),
+            cities_path: optional("CITIES_PATH", "data/cities.json"),
+        })
+    }
+}

src/db.rs

@@ -0,0 +1,202 @@
+use std::sync::Mutex;
+
+use anyhow::Result;
+use jacquard::api::com_atproto::label::Label;
+use jacquard::types::string::{Did, UriValue};
+use rusqlite::{Connection, OptionalExtension, Row, params, params_from_iter};
+
+#[derive(Debug, Clone)]
+pub struct SeqLabel {
+    pub seq: i64,
+    pub label: Label,
+}
+
+pub struct LabelQuery<'a> {
+    /// Exact URIs or prefixes ending in '*'; '*' alone matches all.
+    pub patterns: &'a [&'a str],
+    pub sources: &'a [&'a str],
+    pub cursor: i64,
+    pub limit: i64,
+}
+
+pub struct Db(Mutex<Connection>);
+
+struct LabelRow {
+    seq: i64,
+    src: String,
+    uri: String,
+    val: String,
+    neg: bool,
+    cts: String,
+    sig: Vec<u8>,
+}
+
+const COLUMNS: &str = "seq, src, uri, val, neg, cts, sig";
+
+fn read_row(row: &Row) -> rusqlite::Result<LabelRow> {
+    Ok(LabelRow {
+        seq: row.get(0)?,
+        src: row.get(1)?,
+        uri: row.get(2)?,
+        val: row.get(3)?,
+        neg: row.get(4)?,
+        cts: row.get(5)?,
+        sig: row.get(6)?,
+    })
+}
+
+impl TryFrom<LabelRow> for SeqLabel {
+    type Error = anyhow::Error;
+
+    fn try_from(row: LabelRow) -> Result<Self> {
+        Ok(SeqLabel {
+            seq: row.seq,
+            label: Label {
+                cid: None,
+                cts: row.cts.parse()?,
+                exp: None,
+                neg: row.neg.then_some(true),
+                sig: Some(row.sig.into()),
+                src: Did::new_owned(row.src)?,
+                uri: UriValue::new_owned(row.uri)?,
+                val: row.val.into(),
+                ver: Some(1),
+                extra_data: None,
+            },
+        })
+    }
+}
+
+impl Db {
+    pub fn open(path: &str) -> Result<Self> {
+        let conn = Connection::open(path)?;
+        conn.execute_batch(
+            "PRAGMA journal_mode = WAL;
+            CREATE TABLE IF NOT EXISTS labels (
+              seq INTEGER PRIMARY KEY AUTOINCREMENT,
+              src TEXT NOT NULL,
+              uri TEXT NOT NULL,
+              val TEXT NOT NULL,
+              neg INTEGER NOT NULL DEFAULT 0,
+              cts TEXT NOT NULL,
+              sig BLOB NOT NULL
+            );
+            CREATE INDEX IF NOT EXISTS labels_uri ON labels (uri, val, seq);
+            CREATE TABLE IF NOT EXISTS state (
+              key TEXT PRIMARY KEY,
+              value TEXT NOT NULL
+            );",
+        )?;
+        Ok(Self(Mutex::new(conn)))
+    }
+
+    fn conn(&self) -> std::sync::MutexGuard<'_, Connection> {
+        self.0.lock().expect("db mutex poisoned")
+    }
+
+    pub fn insert_label(&self, label: &Label) -> Result<i64> {
+        let sig = label.sig.as_deref().unwrap_or_default();
+        Ok(self.conn().query_row(
+            "INSERT INTO labels (src, uri, val, neg, cts, sig) VALUES (?, ?, ?, ?, ?, ?) RETURNING seq",
+            params![
+                label.src.as_str(),
+                label.uri.as_str(),
+                label.val.as_str(),
+                label.neg.unwrap_or(false),
+                label.cts.as_str(),
+                sig
+            ],
+            |row| row.get(0),
+        )?)
+    }
+
+    /// Values whose latest row for `uri` is an application, not a negation.
+    pub fn active_vals(&self, uri: &str) -> Result<Vec<String>> {
+        let conn = self.conn();
+        let mut stmt = conn.prepare_cached(
+            "SELECT val FROM labels l
+            WHERE uri = ? AND neg = 0
+              AND seq = (SELECT MAX(seq) FROM labels WHERE uri = l.uri AND val = l.val)",
+        )?;
+        let vals = stmt.query_map([uri], |row| row.get(0))?;
+        Ok(vals.collect::<rusqlite::Result<_>>()?)
+    }
+
+    pub fn labels_after(&self, seq: i64, limit: i64) -> Result<Vec<SeqLabel>> {
+        let conn = self.conn();
+        let mut stmt = conn.prepare_cached(&format!(
+            "SELECT {COLUMNS} FROM labels WHERE seq > ? ORDER BY seq LIMIT ?"
+        ))?;
+        let rows = stmt.query_map(params![seq, limit], read_row)?;
+        rows.map(|r| r?.try_into()).collect()
+    }
+
+    pub fn max_seq(&self) -> Result<i64> {
+        Ok(self
+            .conn()
+            .query_row("SELECT COALESCE(MAX(seq), 0) FROM labels", [], |row| {
+                row.get(0)
+            })?)
+    }
+
+    pub fn query_labels(&self, q: &LabelQuery) -> Result<Vec<SeqLabel>> {
+        let mut clauses = vec!["seq > ?".to_owned()];
+        let mut args: Vec<rusqlite::types::Value> = vec![q.cursor.into()];
+        if !q.patterns.contains(&"*") {
+            let uris: Vec<_> = q
+                .patterns
+                .iter()
+                .map(|p| match p.strip_suffix('*') {
+                    None => {
+                        args.push(p.to_string().into());
+                        "uri = ?"
+                    }
+                    Some(prefix) => {
+                        let escaped = prefix
+                            .replace('\\', "\\\\")
+                            .replace('%', "\\%")
+                            .replace('_', "\\_");
+                        args.push(format!("{escaped}%").into());
+                        "uri LIKE ? ESCAPE '\\'"
+                    }
+                })
+                .collect();
+            clauses.push(format!("({})", uris.join(" OR ")));
+        }
+        if !q.sources.is_empty() {
+            clauses.push(format!(
+                "src IN ({})",
+                vec!["?"; q.sources.len()].join(", ")
+            ));
+            args.extend(q.sources.iter().map(|s| s.to_string().into()));
+        }
+        args.push(q.limit.into());
+
+        let sql = format!(
+            "SELECT {COLUMNS} FROM labels WHERE {} ORDER BY seq LIMIT ?",
+            clauses.join(" AND ")
+        );
+        let conn = self.conn();
+        let mut stmt = conn.prepare(&sql)?;
+        let rows = stmt.query_map(params_from_iter(args), read_row)?;
+        rows.map(|r| r?.try_into()).collect()
+    }
+
+    pub fn get_state(&self, key: &str) -> Result<Option<String>> {
+        Ok(self
+            .conn()
+            .query_row("SELECT value FROM state WHERE key = ?", [key], |row| {
+                row.get(0)
+            })
+            .optional()?)
+    }
+
+    pub fn set_state(&self, key: &str, value: &str) -> Result<()> {
+        self.conn().execute(
+            "INSERT INTO state (key, value) VALUES (?, ?)
+            ON CONFLICT (key) DO UPDATE SET value = excluded.value",
+            [key, value],
+        )?;
+        Ok(())
+    }
+}

src/dms.rs

@@ -0,0 +1,163 @@
+use std::time::Duration;
+
+use anyhow::Result;
+use jacquard::api::app_bsky::graph::get_relationships::{
+    GetRelationships, GetRelationshipsOutputRelationshipsItem,
+};
+use jacquard::api::chat_bsky::convo::accept_convo::AcceptConvo;
+use jacquard::api::chat_bsky::convo::get_log::{GetLog, GetLogOutputLogsItem};
+use jacquard::api::chat_bsky::convo::send_message::SendMessage;
+use jacquard::api::chat_bsky::convo::{LogCreateMessageMessage, MessageInput};
+use jacquard::client::{Agent, MemoryCredentialSession};
+use jacquard::prelude::*;
+use jacquard::types::ident::AtIdentifier;
+use jacquard::types::string::Did;
+use jacquard::xrpc::CallOptions;
+use jacquard::{CowStr, DefaultStr};
+
+use crate::cities::Cities;
+use crate::labeler::Labeler;
+
+const POLL_INTERVAL: Duration = Duration::from_secs(3);
+const CURSOR_KEY: &str = "chat_cursor";
+const CHAT_SERVICE: &str = "did:web:api.bsky.chat#bsky_chat";
+
+const ONBOARDING: &str = "Hi! To get a city badge on your profile:
+1. Subscribe to this labeler (the Subscribe button on its profile), or you won't see anyone's badges
+2. Follow this account
+Then send me your city, e.g. \"Halifax\".";
+
+const HELP: &str =
+    "Send me your city, e.g. \"Halifax\" or \"Richmond, BC\". Send \"remove\" to clear it.";
+
+pub struct Dms<'a> {
+    pub agent: &'a Agent<MemoryCredentialSession>,
+    pub cities: &'a Cities,
+    pub labeler: &'a Labeler,
+}
+
+/// Chat calls go through the PDS, proxied to the Bluesky chat service.
+fn chat() -> CallOptions {
+    CallOptions {
+        atproto_proxy: Some(DefaultStr::new_static(CHAT_SERVICE)),
+        ..Default::default()
+    }
+}
+
+impl Dms<'_> {
+    pub async fn run(&self) {
+        loop {
+            if let Err(err) = self.poll().await {
+                eprintln!("chat poll failed: {err:#}");
+            }
+            tokio::time::sleep(POLL_INTERVAL).await;
+        }
+    }
+
+    async fn poll(&self) -> Result<()> {
+        let db = &self.labeler.db;
+        let cursor = db.get_state(CURSOR_KEY)?;
+        let request = GetLog::builder().maybe_cursor(cursor.as_deref()).build();
+        let response = self.agent.send_with_opts(request, chat()).await?;
+        let output = response.parse::<CowStr<'_>>()?;
+
+        for log in &output.logs {
+            let GetLogOutputLogsItem::LogCreateMessage(log) = log else {
+                continue;
+            };
+            let LogCreateMessageMessage::MessageView(message) = &log.message else {
+                continue;
+            };
+            let sender = message.sender.did.borrow();
+            if sender.as_str() == self.labeler.did.as_str() {
+                continue;
+            }
+            if let Err(err) = self.handle(&log.convo_id, &sender, &message.text).await {
+                eprintln!("failed handling DM from {sender}: {err:#}");
+            }
+        }
+        if let Some(next) = &output.cursor {
+            db.set_state(CURSOR_KEY, next)?;
+        }
+        Ok(())
+    }
+
+    async fn handle(&self, convo_id: &str, sender: &Did<&str>, text: &str) -> Result<()> {
+        // DMs from accounts we don't follow arrive as requests; accepting is idempotent
+        let accept = AcceptConvo::<&str> {
+            convo_id,
+            extra_data: None,
+        };
+        self.agent.send_with_opts(accept, chat()).await?;
+
+        let reply = self.respond(sender, text).await?;
+        let message = MessageInput {
+            text: reply.as_str(),
+            embed: None,
+            facets: None,
+            reply_to: None,
+            extra_data: None,
+        };
+        let send = SendMessage::builder()
+            .convo_id(convo_id)
+            .message(message)
+            .build();
+        self.agent.send_with_opts(send, chat()).await?;
+        Ok(())
+    }
+
+    async fn respond(&self, sender: &Did<&str>, text: &str) -> Result<String> {
+        if !self.follows_us(sender).await? {
+            return Ok(ONBOARDING.to_owned());
+        }
+
+        let command = text.trim().to_lowercase();
+        if command == "remove" {
+            self.labeler.set_city(sender, None)?;
+            return Ok("Removed your city badge.".to_owned());
+        }
+        if command == "help" || command.is_empty() {
+            let current = self.labeler.current_city(sender)?;
+            return Ok(
+                match current.as_deref().and_then(|id| self.cities.get(id)) {
+                    Some(city) => format!("Your city is {}. {HELP}", city.display()),
+                    None => HELP.to_owned(),
+                },
+            );
+        }
+
+        let found = self.cities.find(text);
+        if let [city] = found.exact.as_slice() {
+            self.labeler.set_city(sender, Some(&city.id))?;
+            return Ok(format!("Your badge is now {}.", city.display()));
+        }
+        let options = if found.exact.is_empty() {
+            found.suggestions
+        } else {
+            found.exact
+        };
+        Ok(match options.first() {
+            Some(first) => {
+                let names: Vec<_> = options.iter().map(|c| c.display()).collect();
+                format!(
+                    "Did you mean: {}? Reply with one, e.g. \"{}\".",
+                    names.join(" / "),
+                    first.display()
+                )
+            }
+            None => format!("I don't know \"{}\". {HELP}", text.trim()),
+        })
+    }
+
+    async fn follows_us(&self, did: &Did<&str>) -> Result<bool> {
+        let request = GetRelationships::builder()
+            .actor(AtIdentifier::Did(self.labeler.did.borrow()))
+            .others(vec![AtIdentifier::Did(did.clone())])
+            .build();
+        let response = self.agent.send(request).await?;
+        let output = response.parse::<CowStr<'_>>()?;
+        Ok(output.relationships.iter().any(|r| {
+            matches!(r, GetRelationshipsOutputRelationshipsItem::Relationship(r) if r.followed_by.is_some())
+        }))
+    }
+}

src/labeler.rs

@@ -0,0 +1,61 @@
+use std::sync::Arc;
+
+use anyhow::Result;
+use jacquard::types::string::Did;
+use k256::ecdsa::SigningKey;
+use tokio::sync::broadcast;
+
+use crate::db::{Db, SeqLabel};
+use crate::labels::sign_label;
+
+pub struct Labeler {
+    pub db: Arc<Db>,
+    pub did: Did,
+    key: SigningKey,
+    published: broadcast::Sender<SeqLabel>,
+}
+
+impl Labeler {
+    pub fn new(db: Arc<Db>, did: Did, key: SigningKey) -> Self {
+        let (published, _) = broadcast::channel(1024);
+        Self {
+            db,
+            did,
+            key,
+            published,
+        }
+    }
+
+    /// Live labels for `subscribeLabels`.
+    pub fn subscribe(&self) -> broadcast::Receiver<SeqLabel> {
+        self.published.subscribe()
+    }
+
+    pub fn current_city(&self, subject: &Did<&str>) -> Result<Option<String>> {
+        Ok(self.db.active_vals(subject.as_str())?.into_iter().next())
+    }
+
+    /// Make `city` the account's only city label; `None` removes it.
+    pub fn set_city(&self, subject: &Did<&str>, city: Option<&str>) -> Result<()> {
+        let active = self.db.active_vals(subject.as_str())?;
+        for val in active.iter().filter(|v| Some(v.as_str()) != city) {
+            self.emit(subject, val, true)?;
+        }
+        if let Some(city) = city.filter(|c| !active.iter().any(|v| v == c)) {
+            self.emit(subject, city, false)?;
+        }
+        Ok(())
+    }
+
+    fn emit(&self, subject: &Did<&str>, val: &str, neg: bool) -> Result<()> {
+        let label = sign_label(&self.key, &self.did, subject, val, neg)?;
+        let seq = self.db.insert_label(&label)?;
+        // no receivers just means no one is subscribed right now
+        let _ = self.published.send(SeqLabel { seq, label });
+        println!(
+            "{} {val} on {subject} (seq {seq})",
+            if neg { "negated" } else { "applied" }
+        );
+        Ok(())
+    }
+}

src/labels.rs

@@ -0,0 +1,30 @@
+use anyhow::Result;
+use jacquard::api::com_atproto::label::Label;
+use jacquard::types::string::{Datetime, Did, UriValue};
+use k256::ecdsa::{Signature, SigningKey, signature::Signer};
+
+/// Labels are signed over the DAG-CBOR encoding of the label without `sig`,
+/// with `neg` omitted unless true (matching Ozone).
+pub fn sign_label(
+    key: &SigningKey,
+    src: &Did,
+    uri: &Did<&str>,
+    val: &str,
+    neg: bool,
+) -> Result<Label> {
+    let mut label = Label {
+        cid: None,
+        cts: Datetime::now(),
+        exp: None,
+        neg: neg.then_some(true),
+        sig: None,
+        src: src.clone(),
+        uri: UriValue::new_owned(uri.as_str())?,
+        val: val.into(),
+        ver: Some(1),
+        extra_data: None,
+    };
+    let sig: Signature = key.sign(&serde_ipld_dagcbor::to_vec(&label)?);
+    label.sig = Some(sig.to_bytes().to_vec().into());
+    Ok(label)
+}

src/main.rs

@@ -0,0 +1,56 @@
+mod cities;
+mod config;
+mod db;
+mod dms;
+mod labeler;
+mod labels;
+mod server;
+
+use std::sync::Arc;
+
+use anyhow::{Context, Result};
+use jacquard::client::{Agent, MemoryCredentialSession};
+use k256::ecdsa::SigningKey;
+
+use crate::cities::Cities;
+use crate::config::Config;
+use crate::db::Db;
+use crate::dms::Dms;
+use crate::labeler::Labeler;
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    let config = Config::from_env()?;
+
+    let (session, account) =
+        MemoryCredentialSession::authenticated(&config.handle, &config.app_password, None, None)
+            .await
+            .with_context(|| format!("logging in as {}", config.handle))?;
+    let agent = Agent::from(session);
+
+    let key = SigningKey::from_slice(&hex::decode(&config.signing_key_hex)?)
+        .context("SIGNING_KEY_HEX")?;
+    let cities = Cities::load(&config.cities_path)?;
+    let db = Arc::new(Db::open(&config.db_path)?);
+    let labeler = Arc::new(Labeler::new(db, account.did, key));
+    println!(
+        "labeling as {}, loaded {} cities",
+        labeler.did,
+        cities.len()
+    );
+
+    let listener = tokio::net::TcpListener::bind(("0.0.0.0", config.port)).await?;
+    println!("labeler listening on :{}", config.port);
+    let server = axum::serve(listener, server::router(labeler.clone()));
+
+    let dms = Dms {
+        agent: &agent,
+        cities: &cities,
+        labeler: &labeler,
+    };
+    tokio::select! {
+        result = server => result?,
+        () = dms.run() => {}
+    }
+    Ok(())
+}

src/server.rs

@@ -0,0 +1,180 @@
+use std::sync::Arc;
+
+use anyhow::Result;
+use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade};
+use axum::extract::{Query, State};
+use axum::http::StatusCode;
+use axum::response::Response;
+use axum::routing::get;
+use axum::{Json, Router};
+use jacquard::api::com_atproto::label::query_labels::{QueryLabelsOutput, QueryLabelsRequest};
+use jacquard::api::com_atproto::label::subscribe_labels::{Labels, SubscribeLabels};
+use jacquard_axum::{ExtractXrpc, GenericXrpcErrorResponse, IntoRouter, XrpcResponse};
+use serde::Serialize;
+use tokio::sync::broadcast::error::RecvError;
+
+use crate::db::{LabelQuery, SeqLabel};
+use crate::labeler::Labeler;
+
+const REPLAY_BATCH: i64 = 500;
+
+fn bad_request(message: &str) -> GenericXrpcErrorResponse {
+    GenericXrpcErrorResponse::new(StatusCode::BAD_REQUEST, "InvalidRequest", Some(message))
+}
+
+fn internal_error(err: anyhow::Error) -> GenericXrpcErrorResponse {
+    eprintln!("request failed: {err:#}");
+    GenericXrpcErrorResponse::internal_server_error()
+}
+
+async fn query_labels(
+    State(labeler): State<Arc<Labeler>>,
+    ExtractXrpc(req): ExtractXrpc<QueryLabelsRequest>,
+) -> Result<XrpcResponse<QueryLabelsRequest>, GenericXrpcErrorResponse> {
+    // '*' is only allowed as a trailing wildcard
+    if let Some(bad) = req
+        .uri_patterns
+        .iter()
+        .find(|p| p.find('*').is_some_and(|i| i != p.len() - 1))
+    {
+        return Err(bad_request(&format!("invalid pattern: {bad}")));
+    }
+    let cursor = match &req.cursor {
+        Some(c) => c.parse().map_err(|_| bad_request("invalid cursor"))?,
+        None => 0,
+    };
+    let patterns: Vec<&str> = req.uri_patterns.iter().map(|p| p.as_str()).collect();
+    let sources: Vec<&str> = req.sources.iter().flatten().map(|s| s.as_str()).collect();
+
+    let rows = labeler
+        .db
+        .query_labels(&LabelQuery {
+            patterns: &patterns,
+            sources: &sources,
+            cursor,
+            limit: req.limit.unwrap_or(50).clamp(1, 250),
+        })
+        .map_err(internal_error)?;
+    Ok(XrpcResponse(QueryLabelsOutput {
+        cursor: rows.last().map(|r| r.seq.to_string().into()),
+        labels: rows.into_iter().map(|r| r.label).collect(),
+        extra_data: None,
+    }))
+}
+
+// Event stream frames are two concatenated DAG-CBOR objects: header + body.
+// Header fields are declared in DAG-CBOR key order (shorter keys first).
+#[derive(Serialize)]
+struct MessageHeader {
+    t: &'static str,
+    op: i64,
+}
+
+#[derive(Serialize)]
+struct ErrorHeader {
+    op: i64,
+}
+
+#[derive(Serialize)]
+struct ErrorBody {
+    error: &'static str,
+    message: &'static str,
+}
+
+fn frame(header: &impl Serialize, body: &impl Serialize) -> Result<Message> {
+    let mut bytes = serde_ipld_dagcbor::to_vec(header)?;
+    bytes.extend(serde_ipld_dagcbor::to_vec(body)?);
+    Ok(Message::Binary(bytes.into()))
+}
+
+fn labels_frame(row: SeqLabel) -> Result<Message> {
+    let body = Labels {
+        seq: row.seq,
+        labels: vec![row.label],
+        extra_data: None,
+    };
+    frame(
+        &MessageHeader {
+            t: "#labels",
+            op: 1,
+        },
+        &body,
+    )
+}
+
+async fn subscribe_labels(
+    State(labeler): State<Arc<Labeler>>,
+    Query(params): Query<SubscribeLabels>,
+    ws: WebSocketUpgrade,
+) -> Response {
+    ws.on_upgrade(move |mut socket| async move {
+        if let Err(err) = stream_labels(&mut socket, &labeler, params.cursor).await {
+            eprintln!("subscribeLabels ended: {err:#}");
+        }
+        let _ = socket.send(Message::Close(None)).await;
+    })
+}
+
+async fn stream_labels(
+    socket: &mut WebSocket,
+    labeler: &Labeler,
+    cursor: Option<i64>,
+) -> Result<()> {
+    let current = labeler.db.max_seq()?;
+    if cursor.is_some_and(|c| c > current) {
+        let body = ErrorBody {
+            error: "FutureCursor",
+            message: "Cursor in the future.",
+        };
+        socket.send(frame(&ErrorHeader { op: -1 }, &body)?).await?;
+        return Ok(());
+    }
+
+    // subscribe before replaying so nothing published mid-replay is missed
+    let mut live = labeler.subscribe();
+    let mut last = cursor.unwrap_or(current);
+    if cursor.is_some() {
+        replay(socket, labeler, &mut last).await?;
+    }
+
+    loop {
+        match live.recv().await {
+            Ok(row) if row.seq > last => {
+                last = row.seq;
+                socket.send(labels_frame(row)?).await?;
+            }
+            Ok(_) => {}
+            // fell behind the channel; catch up from the database
+            Err(RecvError::Lagged(_)) => replay(socket, labeler, &mut last).await?,
+            Err(RecvError::Closed) => return Ok(()),
+        }
+    }
+}
+
+async fn replay(socket: &mut WebSocket, labeler: &Labeler, last: &mut i64) -> Result<()> {
+    loop {
+        let batch = labeler.db.labels_after(*last, REPLAY_BATCH)?;
+        let done = (batch.len() as i64) < REPLAY_BATCH;
+        for row in batch {
+            *last = row.seq;
+            socket.send(labels_frame(row)?).await?;
+        }
+        if done {
+            return Ok(());
+        }
+    }
+}
+
+pub fn router(labeler: Arc<Labeler>) -> Router {
+    Router::new()
+        .route(
+            "/xrpc/_health",
+            get(|| async { Json(serde_json::json!({ "status": "ok" })) }),
+        )
+        .route(
+            "/xrpc/com.atproto.label.subscribeLabels",
+            get(subscribe_labels),
+        )
+        .merge(QueryLabelsRequest::into_router(query_labels))
+        .with_state(labeler)
+}