Compare commits
7
Commits
a7d410fd6d
..
main
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
1a08d2efda | ||
|
|
9384e2bcea | ||
|
|
edf19507b4 | ||
|
|
b4ac30c010 | ||
|
|
3c0d56cdaf | ||
|
|
049a198b1c | ||
|
|
d1b01e180d |
No files matched your search
@@ -0,0 +1,4 @@
|
|||||||
|
# only the release binary and the city list go into the image
|
||||||
|
*
|
||||||
|
!target/x86_64-unknown-linux-musl/release/cities-labeler
|
||||||
|
!data
|
||||||
Generated
+23
@@ -461,6 +461,8 @@ dependencies = [
|
|||||||
"jacquard-axum",
|
"jacquard-axum",
|
||||||
"k256",
|
"k256",
|
||||||
"rand_core 0.6.4",
|
"rand_core 0.6.4",
|
||||||
|
"reqwest",
|
||||||
|
"rpassword",
|
||||||
"rusqlite",
|
"rusqlite",
|
||||||
"serde",
|
"serde",
|
||||||
"serde_ipld_dagcbor",
|
"serde_ipld_dagcbor",
|
||||||
@@ -2847,6 +2849,17 @@ dependencies = [
|
|||||||
"windows-sys 0.52.0",
|
"windows-sys 0.52.0",
|
||||||
]
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "rpassword"
|
||||||
|
version = "7.5.4"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "2da316a15f47e3d053de9cb2c439650bd8fa4aaeb9365f2e5f27f492ff73c196"
|
||||||
|
dependencies = [
|
||||||
|
"libc",
|
||||||
|
"rtoolbox",
|
||||||
|
"windows-sys 0.61.2",
|
||||||
|
]
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "rsa"
|
name = "rsa"
|
||||||
version = "0.9.10"
|
version = "0.9.10"
|
||||||
@@ -2867,6 +2880,16 @@ dependencies = [
|
|||||||
"zeroize",
|
"zeroize",
|
||||||
]
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "rtoolbox"
|
||||||
|
version = "0.0.6"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "9a1efe12a1469752d0e6ff5ebec0b6ef4924cc5c4c71046b0ec730040535819d"
|
||||||
|
dependencies = [
|
||||||
|
"libc",
|
||||||
|
"windows-sys 0.61.2",
|
||||||
|
]
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "rusqlite"
|
name = "rusqlite"
|
||||||
version = "0.37.0"
|
version = "0.37.0"
|
||||||
|
|||||||
@@ -14,6 +14,8 @@ jacquard = { version = "=0.12.2", default-features = false, features = ["api_blu
|
|||||||
jacquard-axum = "=0.12.2"
|
jacquard-axum = "=0.12.2"
|
||||||
k256 = { version = "0.13", features = ["ecdsa"] }
|
k256 = { version = "0.13", features = ["ecdsa"] }
|
||||||
rand_core = { version = "0.6", features = ["getrandom"] }
|
rand_core = { version = "0.6", features = ["getrandom"] }
|
||||||
|
reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "json"] }
|
||||||
|
rpassword = "7"
|
||||||
rusqlite = { version = "0.37", features = ["bundled"] }
|
rusqlite = { version = "0.37", features = ["bundled"] }
|
||||||
serde = { version = "1", features = ["derive"] }
|
serde = { version = "1", features = ["derive"] }
|
||||||
serde_ipld_dagcbor = "0.6"
|
serde_ipld_dagcbor = "0.6"
|
||||||
|
|||||||
+12
@@ -0,0 +1,12 @@
|
|||||||
|
# Release image: packages the static binary built by scripts/release.sh.
|
||||||
|
# No compiler runs here, so the image builds in seconds on any machine.
|
||||||
|
FROM gcr.io/distroless/static-debian12:nonroot
|
||||||
|
COPY target/x86_64-unknown-linux-musl/release/cities-labeler /usr/local/bin/cities-labeler
|
||||||
|
COPY data /app/data
|
||||||
|
ENV CITIES_PATH=/app/data/cities.json \
|
||||||
|
DB_PATH=/data/labeler.sqlite \
|
||||||
|
PORT=4100
|
||||||
|
# matches the owner of the ansible-managed data directory
|
||||||
|
USER 1000:1000
|
||||||
|
EXPOSE 4100
|
||||||
|
ENTRYPOINT ["/usr/local/bin/cities-labeler"]
|
||||||
@@ -1,4 +1,4 @@
|
|||||||
# cosocial cities labeler
|
# Canadian Cities Labeler
|
||||||
|
|
||||||
An atproto labeler that puts a Canadian city badge on accounts that ask for one over Bluesky DMs.
|
An atproto labeler that puts a Canadian city badge on accounts that ask for one over Bluesky DMs.
|
||||||
|
|
||||||
@@ -10,17 +10,6 @@ The labeler polls its DMs, matches the city against `data/cities.json`, and sign
|
|||||||
label (e.g. `halifax-ns`), negating any previous one. Labels are stored in SQLite and served over
|
label (e.g. `halifax-ns`), negating any previous one. Labels are stored in SQLite and served over
|
||||||
`com.atproto.label.subscribeLabels` and `queryLabels`.
|
`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.
|
Built on [jacquard](https://tangled.org/nonbinary.computer/jacquard) for atproto types and XRPC.
|
||||||
|
|
||||||
## Local test
|
## Local test
|
||||||
@@ -39,3 +28,32 @@ DM it from a test account, then check
|
|||||||
|
|
||||||
Labels won't appear in the Bluesky app until the labeler is deployed publicly and its 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.
|
document has the `#atproto_label` key and `#atproto_labeler` service.
|
||||||
|
|
||||||
|
## First time setup
|
||||||
|
|
||||||
|
```sh
|
||||||
|
cargo run --bin setup -- declare # label definitions; gives the account a Subscribe button
|
||||||
|
cargo run --bin setup -- did # run after deploying; requires labeler account's main password + emailed code
|
||||||
|
```
|
||||||
|
|
||||||
|
## Release
|
||||||
|
|
||||||
|
Releases are made manually with the Rust toolchain. `scripts/release.sh` cross-compiles a static
|
||||||
|
linux/amd64 binary, packages it into a container image, pushes
|
||||||
|
`git.cosocial.info/cosocial-canada/atproto-cosocial-labeler:<version>` and `:latest`, and tags the
|
||||||
|
commit as `v<version>`. It refuses to run with uncommitted changes or an existing tag.
|
||||||
|
|
||||||
|
Prerequisites (once):
|
||||||
|
|
||||||
|
```sh
|
||||||
|
brew install zig # linker and C compiler for cross-compiling
|
||||||
|
cargo install --locked cargo-zigbuild
|
||||||
|
docker login git.cosocial.info # token with the write:package scope
|
||||||
|
```
|
||||||
|
|
||||||
|
The pinned toolchain and its `x86_64-unknown-linux-musl` target come from `rust-toolchain.toml`.
|
||||||
|
Docker must be running. Then:
|
||||||
|
|
||||||
|
```sh
|
||||||
|
scripts/release.sh 0.1.0
|
||||||
|
```
|
||||||
@@ -2,3 +2,5 @@
|
|||||||
channel = "1.91.0"
|
channel = "1.91.0"
|
||||||
profile = "minimal"
|
profile = "minimal"
|
||||||
components = ["rustfmt", "clippy", "rust-analyzer", "rust-src"]
|
components = ["rustfmt", "clippy", "rust-analyzer", "rust-src"]
|
||||||
|
# release builds cross-compile a static binary (see scripts/release.sh)
|
||||||
|
targets = ["x86_64-unknown-linux-musl"]
|
||||||
Executable
+31
@@ -0,0 +1,31 @@
|
|||||||
|
#!/usr/bin/env bash
|
||||||
|
# Cuts a release: cross-compiles a static linux/amd64 binary, pushes it as a
|
||||||
|
# container image (:<version> and :latest), and tags the commit.
|
||||||
|
#
|
||||||
|
# scripts/release.sh 0.1.0
|
||||||
|
#
|
||||||
|
# Needs: zig + cargo-zigbuild, Docker running, and
|
||||||
|
# `docker login git.cosocial.info` with a write:package token.
|
||||||
|
set -euo pipefail
|
||||||
|
|
||||||
|
version="${1:?usage: scripts/release.sh <version>}"
|
||||||
|
image="git.cosocial.info/cosocial-canada/atproto-cosocial-labeler"
|
||||||
|
target="x86_64-unknown-linux-musl"
|
||||||
|
|
||||||
|
if [[ -n "$(git status --porcelain)" ]]; then
|
||||||
|
echo "working tree is dirty; commit first" >&2
|
||||||
|
exit 1
|
||||||
|
fi
|
||||||
|
if git rev-parse -q --verify "refs/tags/v$version" >/dev/null; then
|
||||||
|
echo "tag v$version already exists" >&2
|
||||||
|
exit 1
|
||||||
|
fi
|
||||||
|
|
||||||
|
cargo zigbuild --release --target "$target" --bin cities-labeler
|
||||||
|
docker build --platform linux/amd64 -t "$image:$version" -t "$image:latest" .
|
||||||
|
docker push "$image:$version"
|
||||||
|
docker push "$image:latest"
|
||||||
|
|
||||||
|
git tag -a "v$version" -m "v$version"
|
||||||
|
git push origin "v$version"
|
||||||
|
echo "released $image:$version"
|
||||||
@@ -0,0 +1,69 @@
|
|||||||
|
//! Publishes the labeler declaration (`app.bsky.labeler.service/self`) with one
|
||||||
|
//! label per city. This is what gives the account a Subscribe button. Safe to
|
||||||
|
//! re-run after editing the city list; it overwrites the record.
|
||||||
|
use anyhow::Result;
|
||||||
|
use cities_labeler::cities::Cities;
|
||||||
|
use cities_labeler::config::Config;
|
||||||
|
use cities_labeler::login;
|
||||||
|
use jacquard::api::app_bsky::labeler::LabelerPolicies;
|
||||||
|
use jacquard::api::app_bsky::labeler::service::Service;
|
||||||
|
use jacquard::api::com_atproto::label::{
|
||||||
|
LabelValue, LabelValueDefinition, LabelValueDefinitionBlurs,
|
||||||
|
LabelValueDefinitionDefaultSetting, LabelValueDefinitionSeverity, LabelValueDefinitionStrings,
|
||||||
|
};
|
||||||
|
use jacquard::prelude::*;
|
||||||
|
use jacquard::types::language::Language;
|
||||||
|
use jacquard::types::recordkey::RecordKey;
|
||||||
|
use jacquard::types::string::Datetime;
|
||||||
|
|
||||||
|
pub async fn run(config: &Config) -> Result<()> {
|
||||||
|
let cities = Cities::load(&config.cities_path)?;
|
||||||
|
let (agent, did) = login(&config.handle, &config.app_password).await?;
|
||||||
|
|
||||||
|
let definitions: Vec<LabelValueDefinition> = cities
|
||||||
|
.iter()
|
||||||
|
.map(|city| LabelValueDefinition {
|
||||||
|
identifier: city.id.as_str().into(),
|
||||||
|
// a neutral badge, shown by default, never hiding content
|
||||||
|
severity: LabelValueDefinitionSeverity::Inform,
|
||||||
|
blurs: LabelValueDefinitionBlurs::None,
|
||||||
|
default_setting: Some(LabelValueDefinitionDefaultSetting::Warn),
|
||||||
|
adult_only: Some(false),
|
||||||
|
locales: vec![LabelValueDefinitionStrings {
|
||||||
|
lang: Language::new_static("en").expect("valid language tag"),
|
||||||
|
name: city.display().into(),
|
||||||
|
description: format!("Chose {} as their city.", city.display()).into(),
|
||||||
|
extra_data: None,
|
||||||
|
}],
|
||||||
|
extra_data: None,
|
||||||
|
})
|
||||||
|
.collect();
|
||||||
|
|
||||||
|
let service: Service = Service {
|
||||||
|
created_at: Datetime::now(),
|
||||||
|
labels: None,
|
||||||
|
policies: LabelerPolicies {
|
||||||
|
label_values: cities
|
||||||
|
.iter()
|
||||||
|
.map(|c| LabelValue::Other(c.id.as_str().into()))
|
||||||
|
.collect(),
|
||||||
|
label_value_definitions: Some(definitions),
|
||||||
|
extra_data: None,
|
||||||
|
},
|
||||||
|
// these declare which reports the labeler reviews: none
|
||||||
|
reason_types: Some(vec![]),
|
||||||
|
subject_types: Some(vec![]),
|
||||||
|
subject_collections: None,
|
||||||
|
extra_data: None,
|
||||||
|
};
|
||||||
|
|
||||||
|
let output = agent
|
||||||
|
.put_record(RecordKey::any_static("self")?, service)
|
||||||
|
.await?;
|
||||||
|
println!(
|
||||||
|
"published {} with {} labels for {did}",
|
||||||
|
output.uri,
|
||||||
|
cities.len()
|
||||||
|
);
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
@@ -0,0 +1,138 @@
|
|||||||
|
//! Adds the label signing key (`#atproto_label`) and the labeler endpoint
|
||||||
|
//! (`#atproto_labeler`) to the account's DID document. Needs the account's main
|
||||||
|
//! password and a code emailed by the PDS. Run it once the labeler is live at
|
||||||
|
//! `LABELER_URL`; it exits early if both entries are already in place.
|
||||||
|
use std::collections::BTreeMap;
|
||||||
|
use std::io::Write;
|
||||||
|
|
||||||
|
use anyhow::{Context, Result, bail};
|
||||||
|
use cities_labeler::config::Config;
|
||||||
|
use cities_labeler::login;
|
||||||
|
use jacquard::AuthorizationToken;
|
||||||
|
use jacquard::api::com_atproto::identity::sign_plc_operation::SignPlcOperation;
|
||||||
|
use jacquard::api::com_atproto::identity::submit_plc_operation::SubmitPlcOperation;
|
||||||
|
use jacquard::client::{Agent, MemoryCredentialSession};
|
||||||
|
use jacquard::common::to_data;
|
||||||
|
use jacquard::common::types::crypto::multikey;
|
||||||
|
use jacquard::prelude::*;
|
||||||
|
use k256::ecdsa::SigningKey;
|
||||||
|
use serde::{Deserialize, Serialize};
|
||||||
|
|
||||||
|
const PLC_DIRECTORY: &str = "https://plc.directory";
|
||||||
|
|
||||||
|
/// The parts of the DID's current PLC state we change; everything else
|
||||||
|
/// (rotation keys, handle) is left as the PDS has it.
|
||||||
|
#[derive(Debug, Deserialize)]
|
||||||
|
#[serde(rename_all = "camelCase")]
|
||||||
|
struct PlcData {
|
||||||
|
verification_methods: BTreeMap<String, String>,
|
||||||
|
services: BTreeMap<String, PlcService>,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, PartialEq, Deserialize, Serialize)]
|
||||||
|
struct PlcService {
|
||||||
|
#[serde(rename = "type")]
|
||||||
|
kind: String,
|
||||||
|
endpoint: String,
|
||||||
|
}
|
||||||
|
|
||||||
|
pub async fn run(config: &Config) -> Result<()> {
|
||||||
|
let key = SigningKey::from_slice(&hex::decode(&config.signing_key_hex)?)
|
||||||
|
.context("SIGNING_KEY_HEX")?;
|
||||||
|
let public = key.verifying_key().to_encoded_point(true);
|
||||||
|
let label_key = format!("did:key:{}", multikey(0xE7, public.as_bytes()));
|
||||||
|
let labeler = PlcService {
|
||||||
|
kind: "AtprotoLabeler".into(),
|
||||||
|
endpoint: config.labeler_url.clone(),
|
||||||
|
};
|
||||||
|
|
||||||
|
// identity changes need the main password; app passwords are refused
|
||||||
|
let password = rpassword::prompt_password(format!("main password for {}: ", config.handle))?;
|
||||||
|
let (agent, did) = login(&config.handle, &password).await?;
|
||||||
|
|
||||||
|
let url = format!("{PLC_DIRECTORY}/{did}/data");
|
||||||
|
let mut plc: PlcData = reqwest::get(&url)
|
||||||
|
.await?
|
||||||
|
.error_for_status()?
|
||||||
|
.json()
|
||||||
|
.await
|
||||||
|
.with_context(|| format!("reading {url}"))?;
|
||||||
|
if plc.verification_methods.get("atproto_label") == Some(&label_key)
|
||||||
|
&& plc.services.get("atproto_labeler") == Some(&labeler)
|
||||||
|
{
|
||||||
|
println!("{did} already points at {}", config.labeler_url);
|
||||||
|
return Ok(());
|
||||||
|
}
|
||||||
|
plc.verification_methods
|
||||||
|
.insert("atproto_label".into(), label_key.clone());
|
||||||
|
plc.services.insert("atproto_labeler".into(), labeler);
|
||||||
|
println!(
|
||||||
|
"will set on {did}:\n #atproto_label {label_key}\n #atproto_labeler {}",
|
||||||
|
config.labeler_url
|
||||||
|
);
|
||||||
|
|
||||||
|
request_signature(&agent).await?;
|
||||||
|
let token = prompt("code emailed to the account: ")?;
|
||||||
|
let signed = agent
|
||||||
|
.send(SignPlcOperation {
|
||||||
|
token: Some(token.as_str().into()),
|
||||||
|
verification_methods: Some(to_data(&plc.verification_methods)?),
|
||||||
|
services: Some(to_data(&plc.services)?),
|
||||||
|
also_known_as: None,
|
||||||
|
rotation_keys: None,
|
||||||
|
extra_data: None,
|
||||||
|
})
|
||||||
|
.await?
|
||||||
|
.into_output()?;
|
||||||
|
|
||||||
|
if prompt("submit this PLC operation? [y/N] ")? != "y" {
|
||||||
|
bail!("aborted; nothing submitted");
|
||||||
|
}
|
||||||
|
let response = agent
|
||||||
|
.send(SubmitPlcOperation {
|
||||||
|
operation: signed.operation,
|
||||||
|
extra_data: None,
|
||||||
|
})
|
||||||
|
.await?;
|
||||||
|
// success is an empty body, which into_output() can't decode as JSON
|
||||||
|
if !response.status().is_success() {
|
||||||
|
response.into_output()?;
|
||||||
|
bail!("submitPlcOperation failed");
|
||||||
|
}
|
||||||
|
println!("submitted; check {PLC_DIRECTORY}/{did}");
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Asks the PDS to email a PLC operation code. Sent by hand because jacquard
|
||||||
|
/// serializes this no-input procedure as a `null` JSON body, which the PDS
|
||||||
|
/// rejects with "A request body was provided when none was expected".
|
||||||
|
async fn request_signature(agent: &Agent<MemoryCredentialSession>) -> Result<()> {
|
||||||
|
let Some(AuthorizationToken::Bearer(token)) = agent.inner().access_token().await else {
|
||||||
|
bail!("not logged in");
|
||||||
|
};
|
||||||
|
let url = format!(
|
||||||
|
"{}/xrpc/com.atproto.identity.requestPlcOperationSignature",
|
||||||
|
agent.endpoint().await.as_str().trim_end_matches('/')
|
||||||
|
);
|
||||||
|
let response = reqwest::Client::new()
|
||||||
|
.post(&url)
|
||||||
|
.bearer_auth(token.as_str())
|
||||||
|
.send()
|
||||||
|
.await?;
|
||||||
|
if !response.status().is_success() {
|
||||||
|
bail!(
|
||||||
|
"requestPlcOperationSignature failed: {} {}",
|
||||||
|
response.status(),
|
||||||
|
response.text().await?
|
||||||
|
);
|
||||||
|
}
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn prompt(label: &str) -> Result<String> {
|
||||||
|
print!("{label}");
|
||||||
|
std::io::stdout().flush()?;
|
||||||
|
let mut line = String::new();
|
||||||
|
std::io::stdin().read_line(&mut line)?;
|
||||||
|
Ok(line.trim().to_owned())
|
||||||
|
}
|
||||||
@@ -0,0 +1,19 @@
|
|||||||
|
//! One-time labeler setup.
|
||||||
|
//!
|
||||||
|
//! - `cargo run --bin setup -- declare` publishes the label definitions.
|
||||||
|
//! - `cargo run --bin setup -- did` publishes the labeler's identity entries.
|
||||||
|
mod declare;
|
||||||
|
mod did;
|
||||||
|
|
||||||
|
use anyhow::{Result, bail};
|
||||||
|
use cities_labeler::config::Config;
|
||||||
|
|
||||||
|
#[tokio::main]
|
||||||
|
async fn main() -> Result<()> {
|
||||||
|
let config = Config::from_env()?;
|
||||||
|
match std::env::args().nth(1).as_deref() {
|
||||||
|
Some("declare") => declare::run(&config).await,
|
||||||
|
Some("did") => did::run(&config).await,
|
||||||
|
_ => bail!("usage: setup <declare|did>"),
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -75,6 +75,14 @@ impl Cities {
|
|||||||
self.keyed.len()
|
self.keyed.len()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pub fn is_empty(&self) -> bool {
|
||||||
|
self.keyed.is_empty()
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn iter(&self) -> impl Iterator<Item = &City> {
|
||||||
|
self.keyed.iter().map(|k| &k.city)
|
||||||
|
}
|
||||||
|
|
||||||
pub fn get(&self, id: &str) -> Option<&City> {
|
pub fn get(&self, id: &str) -> Option<&City> {
|
||||||
self.by_id.get(id).map(|&i| &self.keyed[i].city)
|
self.by_id.get(id).map(|&i| &self.keyed[i].city)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -8,6 +8,8 @@ pub struct Config {
|
|||||||
pub port: u16,
|
pub port: u16,
|
||||||
pub db_path: String,
|
pub db_path: String,
|
||||||
pub cities_path: String,
|
pub cities_path: String,
|
||||||
|
/// Public HTTPS URL where this labeler serves its XRPC endpoints.
|
||||||
|
pub labeler_url: String,
|
||||||
}
|
}
|
||||||
|
|
||||||
fn required(name: &str) -> Result<String> {
|
fn required(name: &str) -> Result<String> {
|
||||||
@@ -28,6 +30,7 @@ impl Config {
|
|||||||
port: optional("PORT", "4100").parse().context("PORT")?,
|
port: optional("PORT", "4100").parse().context("PORT")?,
|
||||||
db_path: optional("DB_PATH", "labeler.sqlite"),
|
db_path: optional("DB_PATH", "labeler.sqlite"),
|
||||||
cities_path: optional("CITIES_PATH", "data/cities.json"),
|
cities_path: optional("CITIES_PATH", "data/cities.json"),
|
||||||
|
labeler_url: optional("LABELER_URL", "https://cities-labeler.cosocial.ca"),
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
+11
-1
@@ -1,4 +1,4 @@
|
|||||||
use std::time::Duration;
|
use std::time::{Duration, Instant};
|
||||||
|
|
||||||
use anyhow::Result;
|
use anyhow::Result;
|
||||||
use jacquard::api::app_bsky::graph::get_relationships::{
|
use jacquard::api::app_bsky::graph::get_relationships::{
|
||||||
@@ -19,6 +19,9 @@ use crate::cities::Cities;
|
|||||||
use crate::labeler::Labeler;
|
use crate::labeler::Labeler;
|
||||||
|
|
||||||
const POLL_INTERVAL: Duration = Duration::from_secs(3);
|
const POLL_INTERVAL: Duration = Duration::from_secs(3);
|
||||||
|
// Access tokens last about two hours. jacquard 0.12.2 only refreshes after a
|
||||||
|
// 401, but the PDS reports expiry as a 400 `ExpiredToken`, so refresh ahead of it.
|
||||||
|
const REFRESH_INTERVAL: Duration = Duration::from_secs(30 * 60);
|
||||||
const CURSOR_KEY: &str = "chat_cursor";
|
const CURSOR_KEY: &str = "chat_cursor";
|
||||||
const CHAT_SERVICE: &str = "did:web:api.bsky.chat#bsky_chat";
|
const CHAT_SERVICE: &str = "did:web:api.bsky.chat#bsky_chat";
|
||||||
|
|
||||||
@@ -46,7 +49,14 @@ fn chat() -> CallOptions {
|
|||||||
|
|
||||||
impl Dms<'_> {
|
impl Dms<'_> {
|
||||||
pub async fn run(&self) {
|
pub async fn run(&self) {
|
||||||
|
let mut last_refresh = Instant::now();
|
||||||
loop {
|
loop {
|
||||||
|
if last_refresh.elapsed() >= REFRESH_INTERVAL {
|
||||||
|
match self.agent.refresh().await {
|
||||||
|
Ok(_) => last_refresh = Instant::now(),
|
||||||
|
Err(err) => eprintln!("session refresh failed: {err:#}"),
|
||||||
|
}
|
||||||
|
}
|
||||||
if let Err(err) = self.poll().await {
|
if let Err(err) = self.poll().await {
|
||||||
eprintln!("chat poll failed: {err:#}");
|
eprintln!("chat poll failed: {err:#}");
|
||||||
}
|
}
|
||||||
|
|||||||
+19
@@ -0,0 +1,19 @@
|
|||||||
|
pub mod cities;
|
||||||
|
pub mod config;
|
||||||
|
pub mod db;
|
||||||
|
pub mod dms;
|
||||||
|
pub mod labeler;
|
||||||
|
pub mod labels;
|
||||||
|
pub mod server;
|
||||||
|
|
||||||
|
use anyhow::{Context, Result};
|
||||||
|
use jacquard::client::{Agent, MemoryCredentialSession};
|
||||||
|
use jacquard::types::string::Did;
|
||||||
|
|
||||||
|
/// Log in as the labeler account with an app password.
|
||||||
|
pub async fn login(handle: &str, password: &str) -> Result<(Agent<MemoryCredentialSession>, Did)> {
|
||||||
|
let (session, account) = MemoryCredentialSession::authenticated(handle, password, None, None)
|
||||||
|
.await
|
||||||
|
.with_context(|| format!("logging in as {handle}"))?;
|
||||||
|
Ok((Agent::from(session), account.did))
|
||||||
|
}
|
||||||
+8
-21
@@ -1,38 +1,25 @@
|
|||||||
mod cities;
|
|
||||||
mod config;
|
|
||||||
mod db;
|
|
||||||
mod dms;
|
|
||||||
mod labeler;
|
|
||||||
mod labels;
|
|
||||||
mod server;
|
|
||||||
|
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
|
|
||||||
use anyhow::{Context, Result};
|
use anyhow::{Context, Result};
|
||||||
use jacquard::client::{Agent, MemoryCredentialSession};
|
use cities_labeler::cities::Cities;
|
||||||
|
use cities_labeler::config::Config;
|
||||||
|
use cities_labeler::db::Db;
|
||||||
|
use cities_labeler::dms::Dms;
|
||||||
|
use cities_labeler::labeler::Labeler;
|
||||||
|
use cities_labeler::{login, server};
|
||||||
use k256::ecdsa::SigningKey;
|
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]
|
#[tokio::main]
|
||||||
async fn main() -> Result<()> {
|
async fn main() -> Result<()> {
|
||||||
let config = Config::from_env()?;
|
let config = Config::from_env()?;
|
||||||
|
|
||||||
let (session, account) =
|
let (agent, did) = login(&config.handle, &config.app_password).await?;
|
||||||
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)?)
|
let key = SigningKey::from_slice(&hex::decode(&config.signing_key_hex)?)
|
||||||
.context("SIGNING_KEY_HEX")?;
|
.context("SIGNING_KEY_HEX")?;
|
||||||
let cities = Cities::load(&config.cities_path)?;
|
let cities = Cities::load(&config.cities_path)?;
|
||||||
let db = Arc::new(Db::open(&config.db_path)?);
|
let db = Arc::new(Db::open(&config.db_path)?);
|
||||||
let labeler = Arc::new(Labeler::new(db, account.did, key));
|
let labeler = Arc::new(Labeler::new(db, did, key));
|
||||||
println!(
|
println!(
|
||||||
"labeling as {}, loaded {} cities",
|
"labeling as {}, loaded {} cities",
|
||||||
labeler.did,
|
labeler.did,
|
||||||
|
|||||||
+14
-1
@@ -1,4 +1,5 @@
|
|||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
|
use std::time::Duration;
|
||||||
|
|
||||||
use anyhow::Result;
|
use anyhow::Result;
|
||||||
use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade};
|
use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade};
|
||||||
@@ -17,6 +18,7 @@ use crate::db::{LabelQuery, SeqLabel};
|
|||||||
use crate::labeler::Labeler;
|
use crate::labeler::Labeler;
|
||||||
|
|
||||||
const REPLAY_BATCH: i64 = 500;
|
const REPLAY_BATCH: i64 = 500;
|
||||||
|
const PING_INTERVAL: Duration = Duration::from_secs(30);
|
||||||
|
|
||||||
fn bad_request(message: &str) -> GenericXrpcErrorResponse {
|
fn bad_request(message: &str) -> GenericXrpcErrorResponse {
|
||||||
GenericXrpcErrorResponse::new(StatusCode::BAD_REQUEST, "InvalidRequest", Some(message))
|
GenericXrpcErrorResponse::new(StatusCode::BAD_REQUEST, "InvalidRequest", Some(message))
|
||||||
@@ -137,8 +139,18 @@ async fn stream_labels(
|
|||||||
replay(socket, labeler, &mut last).await?;
|
replay(socket, labeler, &mut last).await?;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Labels are rare, so also watch the socket: otherwise a subscriber that
|
||||||
|
// hangs up is only noticed on the next send, and its connection leaks.
|
||||||
|
let mut ping = tokio::time::interval(PING_INTERVAL);
|
||||||
loop {
|
loop {
|
||||||
match live.recv().await {
|
tokio::select! {
|
||||||
|
incoming = socket.recv() => match incoming {
|
||||||
|
None | Some(Err(_)) | Some(Ok(Message::Close(_))) => return Ok(()),
|
||||||
|
Some(Ok(_)) => {}
|
||||||
|
},
|
||||||
|
// keeps idle connections from being dropped by peers and proxies
|
||||||
|
_ = ping.tick() => socket.send(Message::Ping(Default::default())).await?,
|
||||||
|
published = live.recv() => match published {
|
||||||
Ok(row) if row.seq > last => {
|
Ok(row) if row.seq > last => {
|
||||||
last = row.seq;
|
last = row.seq;
|
||||||
socket.send(labels_frame(row)?).await?;
|
socket.send(labels_frame(row)?).await?;
|
||||||
@@ -147,6 +159,7 @@ async fn stream_labels(
|
|||||||
// fell behind the channel; catch up from the database
|
// fell behind the channel; catch up from the database
|
||||||
Err(RecvError::Lagged(_)) => replay(socket, labeler, &mut last).await?,
|
Err(RecvError::Lagged(_)) => replay(socket, labeler, &mut last).await?,
|
||||||
Err(RecvError::Closed) => return Ok(()),
|
Err(RecvError::Closed) => return Ok(()),
|
||||||
|
},
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in new issue
Block a user