···

       
28
       
 
       
      - name: get nightly toolchain for jetstream fmt

     

       
29
       
 
       
        run: rustup toolchain install nightly --allow-downgrade -c rustfmt

     

       
30
       
 
       
      - name: fmt

     

       
31
       
-
       
        run: cargo fmt --package links --package constellation --package ufos --package spacedust --package who-am-i -- --check

     

       
32
       
 
       
      - name: fmt jetstream (nightly)

     

       
33
       
 
       
        run: cargo +nightly fmt --package jetstream -- --check

     

       
34
       
 
       
      - name: clippy

     

···

       
28
       
 
       
      - name: get nightly toolchain for jetstream fmt

     

       
29
       
 
       
        run: rustup toolchain install nightly --allow-downgrade -c rustfmt

     

       
30
       
 
       
      - name: fmt

     

       
31
       
+
       
        run: cargo fmt --package links --package constellation --package ufos --package spacedust --package who-am-i --package slingshot -- --check

     

       
32
       
 
       
      - name: fmt jetstream (nightly)

     

       
33
       
 
       
        run: cargo +nightly fmt --package jetstream -- --check

     

       
34
       
 
       
      - name: clippy

     

···

       
5
       
 
       
	cargo test --all-features

     

       
6
       
 
       


     

       
7
       
 
       
fmt:

     

       
8
       
-
       
	cargo fmt --package links --package constellation --package ufos --package spacedust --package who-am-i

     

       
9
       
 
       
	cargo +nightly fmt --package jetstream

     

       
10
       
 
       


     

       
11
       
 
       
clippy:

     

···

       
5
       
 
       
	cargo test --all-features

     

       
6
       
 
       


     

       
7
       
 
       
fmt:

     

       
8
       
+
       
	cargo fmt --package links --package constellation --package ufos --package spacedust --package who-am-i --package slingshot

     

       
9
       
 
       
	cargo +nightly fmt --package jetstream

     

       
10
       
 
       


     

       
11
       
 
       
clippy:

     

···

       
1
       
 
       
use crate::CachedRecord;

     

       
2
       
-
       
use foyer::HybridCache;

     

       
3
       
 
       
use crate::error::ConsumerError;

     

       
0
       
 
       

     

       
4
       
 
       
use jetstream::{

     

       
5
       
 
       
    DefaultJetstreamEndpoints, JetstreamCompression, JetstreamConfig, JetstreamConnector,

     

       
6
       
 
       
    events::{CommitOp, Cursor, EventKind},

     

···

       
1
       
 
       
use crate::CachedRecord;

     

       
0
       
 
       

     

       
2
       
 
       
use crate::error::ConsumerError;

     

       
3
       
+
       
use foyer::HybridCache;

     

       
4
       
 
       
use jetstream::{

     

       
5
       
 
       
    DefaultJetstreamEndpoints, JetstreamCompression, JetstreamConfig, JetstreamConnector,

     

       
6
       
 
       
    events::{CommitOp, Cursor, EventKind},

     

···

       
0
       
 
       

     

       
0
       
 
       

     

       
1
       
 
       
use std::path::Path;

     

       
2
       
-
       
use crate::CachedRecord;

     

       
3
       
-
       
use foyer::{HybridCache, DirectFsDeviceOptions, Engine, HybridCacheBuilder};

     

       
4
       
 
       


     

       
5
       
-
       


     

       
6
       
-
       
pub async fn firehose_cache(dir: impl AsRef<Path>) -> Result<HybridCache<String, CachedRecord>, String> {

     

       
0
       
 
       

     

       
7
       
 
       
    let cache = HybridCacheBuilder::new()

     

       
8
       
 
       
        .with_name("firehose")

     

       
9
       
 
       
        .memory(64 * 2_usize.pow(20))

     

···

       
1
       
+
       
use crate::CachedRecord;

     

       
2
       
+
       
use foyer::{DirectFsDeviceOptions, Engine, HybridCache, HybridCacheBuilder};

     

       
3
       
 
       
use std::path::Path;

     

       
0
       
 
       

     

       
0
       
 
       

     

       
4
       
 
       


     

       
5
       
+
       
pub async fn firehose_cache(

     

       
6
       
+
       
    dir: impl AsRef<Path>,

     

       
7
       
+
       
) -> Result<HybridCache<String, CachedRecord>, String> {

     

       
8
       
 
       
    let cache = HybridCacheBuilder::new()

     

       
9
       
 
       
        .with_name("firehose")

     

       
10
       
 
       
        .memory(64 * 2_usize.pow(20))

     

···

       
6
       
 
       
use clap::Parser;

     

       
7
       
 
       
use tokio_util::sync::CancellationToken;

     

       
8
       
 
       


     

       
9
       
-
       


     

       
10
       
 
       
/// Slingshot record edge cache

     

       
11
       
 
       
#[derive(Parser, Debug, Clone)]

     

       
12
       
 
       
#[command(version, about, long_about = None)]

     
···

       
65
       
 
       
        Ok(())

     

       
66
       
 
       
    });

     

       
67
       
 
       


     

       
68
       
-
       


     

       
69
       
 
       
    tokio::select! {

     

       
70
       
 
       
        _ = shutdown.cancelled() => log::warn!("shutdown requested"),

     

       
71
       
 
       
        Some(r) = tasks.join_next() => {

     
···

       
111
       
 
       
    );

     

       
112
       
 
       
    Ok(())

     

       
113
       
 
       
}

     

       
114
       
-
       


     

···

       
6
       
 
       
use clap::Parser;

     

       
7
       
 
       
use tokio_util::sync::CancellationToken;

     

       
8
       
 
       


     

       
0
       
 
       

     

       
9
       
 
       
/// Slingshot record edge cache

     

       
10
       
 
       
#[derive(Parser, Debug, Clone)]

     

       
11
       
 
       
#[command(version, about, long_about = None)]

     
···

       
64
       
 
       
        Ok(())

     

       
65
       
 
       
    });

     

       
66
       
 
       


     

       
0
       
 
       

     

       
67
       
 
       
    tokio::select! {

     

       
68
       
 
       
        _ = shutdown.cancelled() => log::warn!("shutdown requested"),

     

       
69
       
 
       
        Some(r) = tasks.join_next() => {

     
···

       
109
       
 
       
    );

     

       
110
       
 
       
    Ok(())

     

       
111
       
 
       
}

     

       
0
       
 
       

     

···

       
1
       
-
       
use serde_json::value::RawValue;

     

       
2
       
-
       
use serde::{Serialize, Deserialize};

     

       
3
       
 
       
use jetstream::exports::Cid;

     

       
0
       
 
       

     

       
0
       
 
       

     

       
4
       
 
       


     

       
5
       
 
       
#[derive(Debug, Serialize, Deserialize)]

     

       
6
       
 
       
pub struct RawRecord {

     
···

       
23
       
 
       
    fn from(RawRecord { cid, record }: &RawRecord) -> Self {

     

       
24
       
 
       
        (

     

       
25
       
 
       
            cid.clone(),

     

       
26
       
-
       
            RawValue::from_string(record.to_string()).expect("stored string from RawValue to be valid"),

     

       
0
       
 
       

     

       
27
       
 
       
        )

     

       
28
       
 
       
    }

     

       
29
       
 
       
}

     

···

       
0
       
 
       

     

       
0
       
 
       

     

       
1
       
 
       
use jetstream::exports::Cid;

     

       
2
       
+
       
use serde::{Deserialize, Serialize};

     

       
3
       
+
       
use serde_json::value::RawValue;

     

       
4
       
 
       


     

       
5
       
 
       
#[derive(Debug, Serialize, Deserialize)]

     

       
6
       
 
       
pub struct RawRecord {

     
···

       
23
       
 
       
    fn from(RawRecord { cid, record }: &RawRecord) -> Self {

     

       
24
       
 
       
        (

     

       
25
       
 
       
            cid.clone(),

     

       
26
       
+
       
            RawValue::from_string(record.to_string())

     

       
27
       
+
       
                .expect("stored string from RawValue to be valid"),

     

       
28
       
 
       
        )

     

       
29
       
 
       
    }

     

       
30
       
 
       
}

     

···

       
0
       
 
       

     

       
1
       
 
       
use foyer::HybridCache;

     

       
2
       
-
       
use crate::{error::ServerError, CachedRecord};

     

       
3
       
 
       
use tokio_util::sync::CancellationToken;

     

       
4
       
 
       


     

       
5
       
-
       
use poem::{listener::TcpListener, Route, Server};

     

       
6
       
 
       
use poem_openapi::{

     

       
7
       
-
       
    payload::Json,

     

       
8
       
-
       
    param::Query,

     

       
9
       
-
       
    OpenApi, OpenApiService,

     

       
10
       
-
       
    ApiResponse,

     

       
11
       
-
       
    Object,

     

       
12
       
-
       
    types::Example,

     

       
13
       
 
       
};

     

       
14
       
 
       


     

       
15
       
 
       
fn example_did() -> String {

     
···

       
39
       
 
       
    }

     

       
40
       
 
       
}

     

       
41
       
 
       


     

       
42
       
-
       


     

       
43
       
 
       
fn bad_request_handler(err: poem::Error) -> GetRecordResponse {

     

       
44
       
 
       
    GetRecordResponse::BadRequest(Json(XrpcErrorResponseObject {

     

       
45
       
 
       
        error: "InvalidRequest".to_string(),

     
···

       
63
       
 
       
impl Example for FoundRecordResponseObject {

     

       
64
       
 
       
    fn example() -> Self {

     

       
65
       
 
       
        Self {

     

       
66
       
-
       
            uri: format!("at://{}/{}/{}", example_did(), example_collection(), example_rkey()),

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
67
       
 
       
            cid: Some("bafyreialv3mzvvxaoyrfrwoer3xmabbmdchvrbyhayd7bga47qjbycy74e".to_string()),

     

       
68
       
 
       
            value: serde_json::json!({

     

       
69
       
 
       
                "$type": "app.bsky.feed.like",

     
···

       
130
       
 
       
        cid: Query<Option<String>>,

     

       
131
       
 
       
    ) -> GetRecordResponse {

     

       
132
       
 
       
        // TODO: yeah yeah

     

       
133
       
-
       
        let at_uri = format!(

     

       
134
       
-
       
            "at://{}/{}/{}",

     

       
135
       
-
       
            &*repo, &*collection, &*rkey

     

       
136
       
-
       
        );

     

       
137
       
 
       


     

       
138
       
-
       
        let entry = self.cache

     

       
139
       
-
       
            .fetch(at_uri.clone(), || async move {

     

       
140
       
-
       
                todo!()

     

       
141
       
-
       
            })

     

       
142
       
 
       
            .await

     

       
143
       
 
       
            .unwrap();

     

       
144
       
 
       


     
···

       
151
       
 
       
                if cid.clone().map(|c| c != found_cid).unwrap_or(false) {

     

       
152
       
 
       
                    return GetRecordResponse::BadRequest(Json(XrpcErrorResponseObject {

     

       
153
       
 
       
                        error: "RecordNotFound".to_string(),

     

       
154
       
-
       
                        message: "A record was found but its CID did not match that requested".to_string(),

     

       
0
       
 
       

     

       
155
       
 
       
                    }));

     

       
156
       
 
       
                }

     

       
157
       
 
       
                // TODO: thank u stellz: https://gist.github.com/stella3d/51e679e55b264adff89d00a1e58d0272

     

       
158
       
-
       
                let value = serde_json::from_str(raw_value.get()).expect("RawValue to be valid json");

     

       
0
       
 
       

     

       
159
       
 
       
                GetRecordResponse::Ok(Json(FoundRecordResponseObject {

     

       
160
       
 
       
                    uri: at_uri,

     

       
161
       
 
       
                    cid: Some(found_cid),

     

       
162
       
 
       
                    value,

     

       
163
       
 
       
                }))

     

       
164
       
-
       
            },

     

       
165
       
-
       
            CachedRecord::Deleted => {

     

       
166
       
-
       
                GetRecordResponse::BadRequest(Json(XrpcErrorResponseObject {

     

       
167
       
-
       
                    error: "RecordNotFound".to_string(),

     

       
168
       
-
       
                    message: "This record was deleted".to_string(),

     

       
169
       
-
       
                }))

     

       
170
       
 
       
            }

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
171
       
 
       
        }

     

       
172
       
 
       
    }

     

       
173
       
 
       
}

     
···

       
176
       
 
       
    cache: HybridCache<String, CachedRecord>,

     

       
177
       
 
       
    _shutdown: CancellationToken,

     

       
178
       
 
       
) -> Result<(), ServerError> {

     

       
179
       
-
       
    let api_service =

     

       
180
       
-
       
        OpenApiService::new(Xrpc { cache }, "Slingshot", env!("CARGO_PKG_VERSION"))

     

       
181
       
-
       
            .server("http://localhost:3000")

     

       
182
       
-
       
            .url_prefix("/xrpc");

     

       
183
       
 
       


     

       
184
       
 
       
    let app = Route::new()

     

       
185
       
 
       
        .nest("/", api_service.scalar())

     

···

       
1
       
+
       
use crate::{CachedRecord, error::ServerError};

     

       
2
       
 
       
use foyer::HybridCache;

     

       
0
       
 
       

     

       
3
       
 
       
use tokio_util::sync::CancellationToken;

     

       
4
       
 
       


     

       
5
       
+
       
use poem::{Route, Server, listener::TcpListener};

     

       
6
       
 
       
use poem_openapi::{

     

       
7
       
+
       
    ApiResponse, Object, OpenApi, OpenApiService, param::Query, payload::Json, types::Example,

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
8
       
 
       
};

     

       
9
       
 
       


     

       
10
       
 
       
fn example_did() -> String {

     
···

       
34
       
 
       
    }

     

       
35
       
 
       
}

     

       
36
       
 
       


     

       
0
       
 
       

     

       
37
       
 
       
fn bad_request_handler(err: poem::Error) -> GetRecordResponse {

     

       
38
       
 
       
    GetRecordResponse::BadRequest(Json(XrpcErrorResponseObject {

     

       
39
       
 
       
        error: "InvalidRequest".to_string(),

     
···

       
57
       
 
       
impl Example for FoundRecordResponseObject {

     

       
58
       
 
       
    fn example() -> Self {

     

       
59
       
 
       
        Self {

     

       
60
       
+
       
            uri: format!(

     

       
61
       
+
       
                "at://{}/{}/{}",

     

       
62
       
+
       
                example_did(),

     

       
63
       
+
       
                example_collection(),

     

       
64
       
+
       
                example_rkey()

     

       
65
       
+
       
            ),

     

       
66
       
 
       
            cid: Some("bafyreialv3mzvvxaoyrfrwoer3xmabbmdchvrbyhayd7bga47qjbycy74e".to_string()),

     

       
67
       
 
       
            value: serde_json::json!({

     

       
68
       
 
       
                "$type": "app.bsky.feed.like",

     
···

       
129
       
 
       
        cid: Query<Option<String>>,

     

       
130
       
 
       
    ) -> GetRecordResponse {

     

       
131
       
 
       
        // TODO: yeah yeah

     

       
132
       
+
       
        let at_uri = format!("at://{}/{}/{}", &*repo, &*collection, &*rkey);

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
133
       
 
       


     

       
134
       
+
       
        let entry = self

     

       
135
       
+
       
            .cache

     

       
136
       
+
       
            .fetch(at_uri.clone(), || async move { todo!() })

     

       
0
       
 
       

     

       
137
       
 
       
            .await

     

       
138
       
 
       
            .unwrap();

     

       
139
       
 
       


     
···

       
146
       
 
       
                if cid.clone().map(|c| c != found_cid).unwrap_or(false) {

     

       
147
       
 
       
                    return GetRecordResponse::BadRequest(Json(XrpcErrorResponseObject {

     

       
148
       
 
       
                        error: "RecordNotFound".to_string(),

     

       
149
       
+
       
                        message: "A record was found but its CID did not match that requested"

     

       
150
       
+
       
                            .to_string(),

     

       
151
       
 
       
                    }));

     

       
152
       
 
       
                }

     

       
153
       
 
       
                // TODO: thank u stellz: https://gist.github.com/stella3d/51e679e55b264adff89d00a1e58d0272

     

       
154
       
+
       
                let value =

     

       
155
       
+
       
                    serde_json::from_str(raw_value.get()).expect("RawValue to be valid json");

     

       
156
       
 
       
                GetRecordResponse::Ok(Json(FoundRecordResponseObject {

     

       
157
       
 
       
                    uri: at_uri,

     

       
158
       
 
       
                    cid: Some(found_cid),

     

       
159
       
 
       
                    value,

     

       
160
       
 
       
                }))

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
161
       
 
       
            }

     

       
162
       
+
       
            CachedRecord::Deleted => GetRecordResponse::BadRequest(Json(XrpcErrorResponseObject {

     

       
163
       
+
       
                error: "RecordNotFound".to_string(),

     

       
164
       
+
       
                message: "This record was deleted".to_string(),

     

       
165
       
+
       
            })),

     

       
166
       
 
       
        }

     

       
167
       
 
       
    }

     

       
168
       
 
       
}

     
···

       
171
       
 
       
    cache: HybridCache<String, CachedRecord>,

     

       
172
       
 
       
    _shutdown: CancellationToken,

     

       
173
       
 
       
) -> Result<(), ServerError> {

     

       
174
       
+
       
    let api_service = OpenApiService::new(Xrpc { cache }, "Slingshot", env!("CARGO_PKG_VERSION"))

     

       
175
       
+
       
        .server("http://localhost:3000")

     

       
176
       
+
       
        .url_prefix("/xrpc");

     

       
0
       
 
       

     

       
177
       
 
       


     

       
178
       
 
       
    let app = Route::new()

     

       
179
       
 
       
        .nest("/", api_service.scalar())