prototype sequencer use

Cargo.lock

···
 dependencies = [
  "anyhow",
  "argon2",
- "async-trait",
  "atrium-api 0.25.3",
  "atrium-crypto",
  "atrium-repo",
- "atrium-xrpc",
- "atrium-xrpc-client",
  "axum",
  "azure_core",
  "azure_identity",
···
  "futures",
  "hex",
  "http-cache-reqwest",
- "ipld-core",
- "k256",
- "lazy_static",
  "memmap2",
  "metrics",
  "metrics-exporter-prometheus",
- "multihash 0.19.3",
- "r2d2",
  "rand 0.8.5",
- "regex",
  "reqwest 0.12.15",
  "reqwest-middleware",
  "rsky-common",
···
  "rsky-syntax",
  "secp256k1",
  "serde",
- "serde_bytes",
  "serde_ipld_dagcbor",
- "serde_ipld_dagjson",
  "serde_json",
  "sha2",
  "thiserror 2.0.12",
···
  "ipld-core",
  "scopeguard",
  "serde",
-]
-
-[[package]]
-name = "serde_ipld_dagjson"
-version = "0.2.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3359b47ba7f4a306ef5984665e10539e212e97217afa489437d533208eecda36"
-dependencies = [
- "ipld-core",
- "serde",
- "serde_json",
 ]
 
 [[package]]

Cargo.toml

···
+# cargo-features = ["codegen-backend"]
+
 [package]
 name = "bluepds"
 version = "0.0.0"
···
 
 [profile.dev.package."*"]
 opt-level = 3
+# codegen-backend = "cranelift"
 
 [profile.dev]
 opt-level = 1
+# codegen-backend = "cranelift"
 
 [profile.release]
 opt-level = "s"     # Slightly slows compile times, great improvements to file size and runtime performance.
···
 # expect_used = "deny"
 
 [dependencies]
-multihash = "0.19.3"
+# multihash = "0.19.3"
 diesel = { version = "2.1.5", features = [
     "chrono",
     "sqlite",
···
     "returning_clauses_for_sqlite_3_35",
 ] }
 diesel_migrations = { version = "2.1.0" }
-r2d2 = "0.8.10"
+# r2d2 = "0.8.10"
 
 atrium-repo = "0.1"
 atrium-api = "0.25"
 # atrium-common = { version = "0.1.2", path = "atrium-common" }
 atrium-crypto = "0.1"
 # atrium-identity = { version = "0.1.4", path = "atrium-identity" }
-atrium-xrpc = "0.12"
-atrium-xrpc-client = "0.5"
+# atrium-xrpc = "0.12"
+# atrium-xrpc-client = "0.5"
 # bsky-sdk = { version = "0.1.19", path = "bsky-sdk" }
 rsky-syntax = { git = "https://github.com/blacksky-algorithms/rsky.git" }
 rsky-repo = { git = "https://github.com/blacksky-algorithms/rsky.git" }
···
 # async-stream = "0.3"
 
 # DAG-CBOR codec
-ipld-core = "0.4.2"
+# ipld-core = "0.4.2"
 serde_ipld_dagcbor = { version = "0.6.2", default-features = false, features = [
     "std",
 ] }
-serde_ipld_dagjson = "0.2.0"
+# serde_ipld_dagjson = "0.2.0"
 cidv10 = { version = "0.10.1", package = "cid" }
 
 # Parsing and validation
···
 hex = "0.4.3"
 # langtag = "0.3"
 # multibase = "0.9.1"
-regex = "1.11.1"
+# regex = "1.11.1"
 serde = { version = "1.0.218", features = ["derive"] }
-serde_bytes = "0.11.17"
+# serde_bytes = "0.11.17"
 # serde_html_form = "0.2.6"
 serde_json = "1.0.139"
 # unsigned-varint = "0.8"
···
 # elliptic-curve = "0.13.6"
 # jose-jwa = "0.1.2"
 # jose-jwk = { version = "0.1.2", default-features = false }
-k256 = "0.13.4"
+# k256 = "0.13.4"
 # p256 = { version = "0.13.2", default-features = false }
 rand = "0.8.5"
 sha2 = "0.10.8"
···
 url = "2.5.4"
 uuid = { version = "1.14.0", features = ["v4"] }
 urlencoding = "2.1.3"
-async-trait = "0.1.88"
-lazy_static = "1.5.0"
+# lazy_static = "1.5.0"
 secp256k1 = "0.28.2"
 dotenvy = "0.15.7"
 deadpool-diesel = { version = "0.6.1", features = [

src/account_manager/mod.rs

···
 use std::collections::BTreeMap;
 use std::env;
 use std::time::SystemTime;
+use tokio::sync::RwLock;
 
 pub(crate) mod helpers {
     pub mod account;
···
         email_token::create_email_token(did, purpose, &self.db).await
     }
 }
+
+pub struct SharedAccountManager {
+    pub account_manager: RwLock<AccountManagerCreator>,
+}

src/lib.rs

···
 #[cfg(test)]
 mod tests;
 
+use account_manager::{AccountManager, SharedAccountManager};
 use anyhow::{Context as _, anyhow};
 use atrium_api::types::string::Did;
 use atrium_crypto::keypair::{Export as _, Secp256k1Keypair};
···
 use firehose::FirehoseProducer;
 use http_cache_reqwest::{CacheMode, HttpCacheOptions, MokaManager};
 use rand::Rng as _;
+use rsky_pds::{crawlers::Crawlers, sequencer::Sequencer};
 use serde::{Deserialize, Serialize};
 use service_proxy::service_proxy;
 use std::{
···
     str::FromStr as _,
     sync::Arc,
 };
-use tokio::net::TcpListener;
+use tokio::{net::TcpListener, sync::RwLock};
 use tower_http::{cors::CorsLayer, trace::TraceLayer};
 use tracing::{info, warn};
 use uuid::Uuid;
···
 pub type Result<T> = std::result::Result<T, Error>;
 /// The reqwest client type with middleware.
 pub type Client = reqwest_middleware::ClientWithMiddleware;
-/// The Azure credential type.
-pub type Cred = Arc<dyn TokenCredential>;
+
+/// The Shared Sequencer which requests crawls from upstream relays and emits events to the firehose.
+pub struct SharedSequencer {
+    /// The sequencer instance.
+    pub sequencer: RwLock<Sequencer>,
+}
 
 #[expect(
     clippy::arbitrary_source_item_ordering,
···
     pub verbosity: Verbosity<InfoLevel>,
 }
 
+/// The actor pools for the database connections.
 pub struct ActorPools {
+    /// The database connection pool for the actor's repository.
     pub repo: Pool,
+    /// The database connection pool for the actor's blobs.
     pub blob: Pool,
 }
 
···
 pub struct AppState {
     /// The application configuration.
     pub config: AppConfig,
-    /// The Azure credential.
-    pub cred: Cred,
     /// The main database connection pool. Used for common PDS data, like invite codes.
     pub db: Pool,
     /// Actor-specific database connection pools. Hashed by DID.
···
     /// The simple HTTP client.
     pub simple_client: reqwest::Client,
     /// The firehose producer.
-    pub firehose: FirehoseProducer,
+    pub sequencer: Arc<SharedSequencer>,
+    /// The account manager.
+    pub account_manager: Arc<SharedAccountManager>,
 
     /// The signing key.
     pub signing_key: SigningKey,
···
     // conn.run_pending_migrations(MIGRATIONS)
     //     .expect("should be able to run migrations");
 
-    let (_fh, fhp) = firehose::spawn(client.clone(), config.clone());
+    let hostname = config.host_name.clone();
+    let crawlers: Vec<String> = config
+        .firehose
+        .relays
+        .iter()
+        .map(|s| s.to_string())
+        .collect();
+    let sequencer = Arc::new(SharedSequencer {
+        sequencer: RwLock::new(Sequencer::new(
+            Crawlers::new(hostname, crawlers.clone()),
+            None,
+        )),
+    });
+    let account_manager = SharedAccountManager {
+        account_manager: RwLock::new(AccountManager::creator()),
+    };
 
     let addr = config
         .listen_address
···
         .layer(CorsLayer::permissive())
         .layer(TraceLayer::new_for_http())
         .with_state(AppState {
-            cred,
             config: config.clone(),
             db: pool.clone(),
             db_actors: actor_pools.clone(),
             client: client.clone(),
             simple_client,
-            firehose: fhp,
+            sequencer: sequencer.clone(),
+            account_manager: Arc::new(account_manager),
             signing_key: skey,
             rotation_key: rkey,
         });
···
 
     let result = conn.interact(move |conn| {
         diesel::sql_query(
-            "SELECT (SELECT COUNT(*) FROM accounts) + (SELECT COUNT(*) FROM invites) AS total_count",
+            "SELECT (SELECT COUNT(*) FROM account) + (SELECT COUNT(*) FROM invite_code) AS total_count",
         )
         .get_result::<TotalCount>(conn)
     })
···
     if c == 0 {
         let uuid = Uuid::new_v4().to_string();
 
+        use crate::models::pds as models;
+        use crate::schema::pds::invite_code::dsl as InviteCode;
         let uuid_clone = uuid.clone();
-        _ = conn
-            .interact(move |conn| {
-                diesel::sql_query(
-            "INSERT INTO invites (id, did, count, created_at) VALUES (?, NULL, 1, datetime('now'))",
-        )
-        .bind::<diesel::sql_types::Text, _>(uuid_clone)
-        .execute(conn)
-        .context("failed to create new invite code")
-        .expect("should be able to create invite code")
+        drop(
+            conn.interact(move |conn| {
+                diesel::insert_into(InviteCode::invite_code)
+                    .values(models::InviteCode {
+                        code: uuid_clone,
+                        available_uses: 1,
+                        disabled: 0,
+                        for_account: "None".to_owned(),
+                        created_by: "None".to_owned(),
+                        created_at: "None".to_owned(),
+                    })
+                    .execute(conn)
+                    .context("failed to create new invite code")
             })
             .await
-            .expect("should be able to create invite code");
+            .expect("should be able to create invite code"),
+        );
 
         // N.B: This is a sensitive message, so we're bypassing `tracing` here and
         // logging it directly to console.
···
     });
 
     // Now that the app is live, request a crawl from upstream relays.
-    firehose::reconnect_relays(&client, &config).await;
+    let mut background_sequencer = sequencer.sequencer.write().await.clone();
+    drop(tokio::spawn(
+        async move { background_sequencer.start().await },
+    ));
 
     serve
         .await