Update module privacy (#133)

This squashed commit rolls up a series of changes designed to improve
Flodgatt's public API/module boundary.  Specifically, this limits the
number of Items that are exported outside the top-level modules (in some
cases because they were already not needed outside that module due to
the earlier code reorganization and in some cases by using public
re-exports to export particular Items from a private module).

Similarly, this commit moves the `Event` struct to the `response`
module (maintaining privacy for the `Event`'s implementation details)
while re-exporting the `Id` struct that `Event` uses internally at the
top level.

All of these changes are made with the goal of making Flodgatt's code
easier to reason about in isolation, which should both make it easier to
maintain and make it easier for new contributors to make changes without
understanding the entire codebase.  Additionally, having fewer public
modules will make documenting Flodgatt more extensively much easier.
This commit is contained in:
Daniel Sockwell 2020-04-21 16:39:31 -04:00 committed by GitHub
parent 10fa24c5d3
commit 1dcddc23de
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
23 changed files with 158 additions and 184 deletions

View File

@ -1,4 +1,7 @@
pub use {deployment_cfg::Deployment, postgres_cfg::Postgres, redis_cfg::Redis}; pub(crate) use postgres_cfg::Postgres;
pub(crate) use redis_cfg::Redis;
use deployment_cfg::Deployment;
use self::environmental_variables::EnvVar; use self::environmental_variables::EnvVar;
use super::err::FatalErr; use super::err::FatalErr;

View File

@ -3,27 +3,23 @@ use crate::err::FatalErr;
#[derive(Debug, Default)] #[derive(Debug, Default)]
pub struct Deployment<'a> { pub struct Deployment<'a> {
pub env: Env, pub(crate) env: Env,
pub log_level: LogLevel, pub(crate) log_level: LogLevel,
pub address: FlodgattAddr, pub address: FlodgattAddr,
pub port: Port, pub port: Port,
pub unix_socket: Socket, pub unix_socket: Socket,
pub cors: Cors<'a>, pub cors: Cors<'a>,
pub sse_interval: SseInterval,
pub ws_interval: WsInterval,
pub whitelist_mode: WhitelistMode, pub whitelist_mode: WhitelistMode,
} }
impl Deployment<'_> { impl Deployment<'_> {
pub fn from_env(env: &EnvVar) -> Result<Self, FatalErr> { pub(crate) fn from_env(env: &EnvVar) -> Result<Self, FatalErr> {
let mut cfg = Self { let mut cfg = Self {
env: Env::default().maybe_update(env.get("NODE_ENV"))?, env: Env::default().maybe_update(env.get("NODE_ENV"))?,
log_level: LogLevel::default().maybe_update(env.get("RUST_LOG"))?, log_level: LogLevel::default().maybe_update(env.get("RUST_LOG"))?,
address: FlodgattAddr::default().maybe_update(env.get("BIND"))?, address: FlodgattAddr::default().maybe_update(env.get("BIND"))?,
port: Port::default().maybe_update(env.get("PORT"))?, port: Port::default().maybe_update(env.get("PORT"))?,
unix_socket: Socket::default().maybe_update(env.get("SOCKET"))?, unix_socket: Socket::default().maybe_update(env.get("SOCKET"))?,
sse_interval: SseInterval::default().maybe_update(env.get("SSE_FREQ"))?,
ws_interval: WsInterval::default().maybe_update(env.get("WS_FREQ"))?,
whitelist_mode: WhitelistMode::default().maybe_update(env.get("WHITELIST_MODE"))?, whitelist_mode: WhitelistMode::default().maybe_update(env.get("WHITELIST_MODE"))?,
cors: Cors::default(), cors: Cors::default(),
}; };

View File

@ -1,10 +1,7 @@
use crate::from_env_var; use crate::from_env_var;
use std::{ use std::fmt;
fmt, use std::net::{IpAddr, Ipv4Addr};
net::{IpAddr, Ipv4Addr}, use std::str::FromStr;
str::FromStr,
time::Duration,
};
use strum_macros::{EnumString, EnumVariantNames}; use strum_macros::{EnumString, EnumVariantNames};
from_env_var!( from_env_var!(
@ -38,20 +35,6 @@ from_env_var!(
let (env_var, allowed_values) = ("SOCKET", "any string"); let (env_var, allowed_values) = ("SOCKET", "any string");
let from_str = |s| Some(Some(s.to_string())); let from_str = |s| Some(Some(s.to_string()));
); );
from_env_var!(
/// The time between replies sent via WebSocket
let name = WsInterval;
let default: Duration = Duration::from_millis(100);
let (env_var, allowed_values) = ("WS_FREQ", "a number of milliseconds");
let from_str = |s| s.parse().map(Duration::from_millis).ok();
);
from_env_var!(
/// The time between replies sent via Server Sent Events
let name = SseInterval;
let default: Duration = Duration::from_millis(100);
let (env_var, allowed_values) = ("WS_FREQ", "a number of milliseconds");
let from_str = |s| s.parse().map(Duration::from_millis).ok();
);
from_env_var!( from_env_var!(
/// The port to run Flodgatt on /// The port to run Flodgatt on
let name = Port; let name = Port;

View File

@ -2,7 +2,7 @@ use hashbrown::HashMap;
use std::fmt; use std::fmt;
#[derive(Debug)] #[derive(Debug)]
pub struct EnvVar(pub HashMap<String, String>); pub(crate) struct EnvVar(pub HashMap<String, String>);
impl std::ops::Deref for EnvVar { impl std::ops::Deref for EnvVar {
type Target = HashMap<String, String>; type Target = HashMap<String, String>;
fn deref(&self) -> &HashMap<String, String> { fn deref(&self) -> &HashMap<String, String> {
@ -16,11 +16,11 @@ impl Clone for EnvVar {
} }
} }
impl EnvVar { impl EnvVar {
pub fn new(vars: HashMap<String, String>) -> Self { pub(crate) fn new(vars: HashMap<String, String>) -> Self {
Self(vars) Self(vars)
} }
pub fn maybe_add_env_var(&mut self, key: &str, maybe_value: Option<impl ToString>) { pub(crate) fn maybe_add_env_var(&mut self, key: &str, maybe_value: Option<impl ToString>) {
if let Some(value) = maybe_value { if let Some(value) = maybe_value {
self.0.insert(key.to_string(), value.to_string()); self.0.insert(key.to_string(), value.to_string());
} }
@ -63,7 +63,7 @@ impl fmt::Display for EnvVar {
#[macro_export] #[macro_export]
macro_rules! maybe_update { macro_rules! maybe_update {
($name:ident; $item: tt:$type:ty) => ( ($name:ident; $item: tt:$type:ty) => (
pub fn $name(self, item: Option<$type>) -> Self { pub(crate) fn $name(self, item: Option<$type>) -> Self {
match item { match item {
Some($item) => Self{ $item, ..self }, Some($item) => Self{ $item, ..self },
None => Self { ..self } None => Self { ..self }
@ -106,7 +106,10 @@ macro_rules! from_env_var {
fn inner_from_str($arg: &str) -> Option<$type> { fn inner_from_str($arg: &str) -> Option<$type> {
$body $body
} }
pub fn maybe_update(self, var: Option<&String>) -> Result<Self, crate::err::FatalErr> { pub(crate) fn maybe_update(
self,
var: Option<&String>,
) -> Result<Self, crate::err::FatalErr> {
Ok(match var { Ok(match var {
Some(empty_string) if empty_string.is_empty() => Self::default(), Some(empty_string) if empty_string.is_empty() => Self::default(),
Some(value) => Self(Self::inner_from_str(value).ok_or_else(|| { Some(value) => Self(Self::inner_from_str(value).ok_or_else(|| {

View File

@ -8,12 +8,12 @@ type Result<T> = std::result::Result<T, FatalErr>;
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
pub struct Postgres { pub struct Postgres {
pub user: PgUser, pub(crate) user: PgUser,
pub host: PgHost, pub(crate) host: PgHost,
pub password: PgPass, pub(crate) password: PgPass,
pub database: PgDatabase, pub(crate) database: PgDatabase,
pub port: PgPort, pub(crate) port: PgPort,
pub ssl_mode: PgSslMode, pub(crate) ssl_mode: PgSslMode,
} }
impl EnvVar { impl EnvVar {
@ -51,7 +51,7 @@ impl EnvVar {
impl Postgres { impl Postgres {
/// Configure Postgres and return a connection /// Configure Postgres and return a connection
pub fn from_env(env: EnvVar) -> Result<Self> { pub(crate) fn from_env(env: EnvVar) -> Result<Self> {
let env = match env.get("DATABASE_URL").cloned() { let env = match env.get("DATABASE_URL").cloned() {
Some(url_str) => env.update_with_postgres_url(&url_str)?, Some(url_str) => env.update_with_postgres_url(&url_str)?,
None => env, None => env,

View File

@ -8,12 +8,12 @@ type Result<T> = std::result::Result<T, FatalErr>;
#[derive(Debug, Default)] #[derive(Debug, Default)]
pub struct Redis { pub struct Redis {
pub user: RedisUser, pub(crate) user: RedisUser,
pub password: RedisPass, pub(crate) password: RedisPass,
pub port: RedisPort, pub(crate) port: RedisPort,
pub host: RedisHost, pub(crate) host: RedisHost,
pub db: RedisDb, pub(crate) db: RedisDb,
pub namespace: RedisNamespace, pub(crate) namespace: RedisNamespace,
// **NOTE**: Polling Redis is much more time consuming than polling the `Receiver` (~1ms // **NOTE**: Polling Redis is much more time consuming than polling the `Receiver` (~1ms
// compared to ~50μs). Thus, changing this setting with REDIS_POLL_INTERVAL may be a good // compared to ~50μs). Thus, changing this setting with REDIS_POLL_INTERVAL may be a good
// place to start for performance improvements at the cost of delaying all updates. // place to start for performance improvements at the cost of delaying all updates.
@ -50,7 +50,7 @@ impl Redis {
const DB_SET_WARNING: &'static str = r"Redis database specified, but PubSub connections do not use databases. const DB_SET_WARNING: &'static str = r"Redis database specified, but PubSub connections do not use databases.
For similar functionality, you may wish to set a REDIS_NAMESPACE"; For similar functionality, you may wish to set a REDIS_NAMESPACE";
pub fn from_env(env: EnvVar) -> Result<Self> { pub(crate) fn from_env(env: EnvVar) -> Result<Self> {
let env = match env.get("REDIS_URL").cloned() { let env = match env.get("REDIS_URL").cloned() {
Some(url_str) => env.update_with_redis_url(&url_str)?, Some(url_str) => env.update_with_redis_url(&url_str)?,
None => env, None => env,

View File

@ -2,9 +2,9 @@ mod checked_event;
mod dynamic_event; mod dynamic_event;
mod err; mod err;
pub use checked_event::{CheckedEvent, Id}; pub(crate) use checked_event::{CheckedEvent, Id};
pub use dynamic_event::{DynEvent, DynStatus, EventKind}; pub(crate) use dynamic_event::{DynEvent, EventKind};
pub use err::EventErr; pub(crate) use err::EventErr;
use serde::Serialize; use serde::Serialize;
use std::convert::TryFrom; use std::convert::TryFrom;
@ -19,7 +19,7 @@ pub enum Event {
} }
impl Event { impl Event {
pub fn to_json_string(&self) -> String { pub(crate) fn to_json_string(&self) -> String {
if let Event::Ping = self { if let Event::Ping = self {
"{}".to_string() "{}".to_string()
} else { } else {
@ -32,7 +32,7 @@ impl Event {
} }
} }
pub fn to_warp_reply(&self) -> Option<(impl ServerSentEvent, impl ServerSentEvent)> { pub(crate) fn to_warp_reply(&self) -> Option<(impl ServerSentEvent, impl ServerSentEvent)> {
if let Event::Ping = self { if let Event::Ping = self {
None None
} else { } else {
@ -103,8 +103,8 @@ impl TryFrom<&str> for Event {
Forwarding Redis payload without type checking it.", Forwarding Redis payload without type checking it.",
e e
); );
let dyn_event: DynEvent = serde_json::from_str(&event_txt)?;
Ok(Event::Dynamic(serde_json::from_str(&event_txt)?)) Ok(Event::Dynamic(dyn_event.set_update()?))
} }
} }
} }

View File

@ -11,12 +11,12 @@ mod status;
mod tag; mod tag;
mod visibility; mod visibility;
pub use announcement::Announcement; use announcement::Announcement;
pub(in crate::event) use announcement_reaction::AnnouncementReaction; pub(in crate::event) use announcement_reaction::AnnouncementReaction;
pub use conversation::Conversation; use conversation::Conversation;
pub use id::Id; pub(crate) use id::Id;
pub use notification::Notification; use notification::Notification;
pub use status::Status; use status::Status;
use serde::Deserialize; use serde::Deserialize;

View File

@ -3,10 +3,16 @@ mod attachment;
mod card; mod card;
mod poll; mod poll;
use super::{ use super::account::Account;
account::Account, emoji::Emoji, id::Id, mention::Mention, tag::Tag, visibility::Visibility, use super::emoji::Emoji;
}; use super::id::Id;
use {application::Application, attachment::Attachment, card::Card, poll::Poll}; use super::mention::Mention;
use super::tag::Tag;
use super::visibility::Visibility;
use application::Application;
use attachment::Attachment;
use card::Card;
use poll::Poll;
use crate::request::Blocks; use crate::request::Blocks;
@ -53,7 +59,7 @@ pub struct Status {
impl Status { impl Status {
/// Returns `true` if the status is filtered out based on its language /// Returns `true` if the status is filtered out based on its language
pub fn language_not(&self, allowed_langs: &HashSet<String>) -> bool { pub(crate) fn language_not(&self, allowed_langs: &HashSet<String>) -> bool {
const ALLOW: bool = false; const ALLOW: bool = false;
const REJECT: bool = true; const REJECT: bool = true;
@ -84,7 +90,7 @@ impl Status {
/// * Wrote this toot /// * Wrote this toot
/// * Wrote a toot that this toot is replying to (if any) /// * Wrote a toot that this toot is replying to (if any)
/// * Wrote the toot that this toot is boosting (if any) /// * Wrote the toot that this toot is boosting (if any)
pub fn involves_any(&self, blocks: &Blocks) -> bool { pub(crate) fn involves_any(&self, blocks: &Blocks) -> bool {
const ALLOW: bool = false; const ALLOW: bool = false;
const REJECT: bool = true; const REJECT: bool = true;
let Blocks { let Blocks {

View File

@ -10,14 +10,14 @@ use serde_json::Value;
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)] #[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
pub struct DynEvent { pub struct DynEvent {
#[serde(skip)] #[serde(skip)]
pub kind: EventKind, pub(crate) kind: EventKind,
pub event: String, pub(crate) event: String,
pub payload: Value, pub(crate) payload: Value,
pub queued_at: Option<i64>, pub(crate) queued_at: Option<i64>,
} }
#[derive(Debug, Clone, PartialEq)] #[derive(Debug, Clone, PartialEq)]
pub enum EventKind { pub(crate) enum EventKind {
Update(DynStatus), Update(DynStatus),
NonUpdate, NonUpdate,
} }
@ -29,19 +29,19 @@ impl Default for EventKind {
} }
#[derive(Debug, Clone, PartialEq)] #[derive(Debug, Clone, PartialEq)]
pub struct DynStatus { pub(crate) struct DynStatus {
pub id: Id, pub(crate) id: Id,
pub username: String, pub(crate) username: String,
pub language: Option<String>, pub(crate) language: Option<String>,
pub mentioned_users: HashSet<Id>, pub(crate) mentioned_users: HashSet<Id>,
pub replied_to_user: Option<Id>, pub(crate) replied_to_user: Option<Id>,
pub boosted_user: Option<Id>, pub(crate) boosted_user: Option<Id>,
} }
type Result<T> = std::result::Result<T, EventErr>; type Result<T> = std::result::Result<T, EventErr>;
impl DynEvent { impl DynEvent {
pub fn set_update(self) -> Result<Self> { pub(crate) fn set_update(self) -> Result<Self> {
if self.event == "update" { if self.event == "update" {
let kind = EventKind::Update(DynStatus::new(&self.payload.clone())?); let kind = EventKind::Update(DynStatus::new(&self.payload.clone())?);
Ok(Self { kind, ..self }) Ok(Self { kind, ..self })
@ -52,7 +52,7 @@ impl DynEvent {
} }
impl DynStatus { impl DynStatus {
pub fn new(payload: &Value) -> Result<Self> { pub(crate) fn new(payload: &Value) -> Result<Self> {
use EventErr::*; use EventErr::*;
Ok(Self { Ok(Self {
@ -68,7 +68,7 @@ impl DynStatus {
}) })
} }
/// Returns `true` if the status is filtered out based on its language /// Returns `true` if the status is filtered out based on its language
pub fn language_not(&self, allowed_langs: &HashSet<String>) -> bool { pub(crate) fn language_not(&self, allowed_langs: &HashSet<String>) -> bool {
const ALLOW: bool = false; const ALLOW: bool = false;
const REJECT: bool = true; const REJECT: bool = true;
@ -93,7 +93,7 @@ impl DynStatus {
/// * Wrote this toot /// * Wrote this toot
/// * Wrote a toot that this toot is replying to (if any) /// * Wrote a toot that this toot is replying to (if any)
/// * Wrote the toot that this toot is boosting (if any) /// * Wrote the toot that this toot is boosting (if any)
pub fn involves_any(&self, blocks: &Blocks) -> bool { pub(crate) fn involves_any(&self, blocks: &Blocks) -> bool {
const ALLOW: bool = false; const ALLOW: bool = false;
const REJECT: bool = true; const REJECT: bool = true;
let Blocks { let Blocks {
@ -112,7 +112,6 @@ impl DynStatus {
} }
} }
// involved_users = mentioned_users + author + replied-to user + boosted user
fn involves(&self, blocked_users: &HashSet<Id>) -> bool { fn involves(&self, blocked_users: &HashSet<Id>) -> bool {
// mentions // mentions
let mut involved_users: HashSet<Id> = self.mentioned_users.clone(); let mut involved_users: HashSet<Id> = self.mentioned_users.clone();

View File

@ -6,11 +6,13 @@ pub mod timeline;
mod err; mod err;
mod subscription; mod subscription;
pub use self::err::RequestErr; pub(crate) use self::err::RequestErr;
pub use self::postgres::PgPool; pub(crate) use self::postgres::PgPool;
// TODO consider whether we can remove `Stream` from public API
pub use subscription::{Blocks, Subscription}; pub(crate) use subscription::Blocks;
pub use timeline::{Content, Reach, Stream, Timeline, TimelineErr}; pub use subscription::Subscription;
pub use timeline::Timeline;
pub(crate) use timeline::{Content, Reach, Stream, TimelineErr};
use self::query::Query; use self::query::Query;
use crate::config; use crate::config;

View File

@ -1,7 +1,6 @@
use std::fmt; use std::fmt;
#[derive(Debug)] #[derive(Debug)]
pub enum RequestErr { pub enum RequestErr {
Unknown,
PgPool(r2d2::Error), PgPool(r2d2::Error),
Pg(postgres::Error), Pg(postgres::Error),
} }
@ -12,7 +11,6 @@ impl fmt::Display for RequestErr {
fn fmt(&self, f: &mut fmt::Formatter) -> Result<(), fmt::Error> { fn fmt(&self, f: &mut fmt::Formatter) -> Result<(), fmt::Error> {
use RequestErr::*; use RequestErr::*;
let msg = match self { let msg = match self {
Unknown => "Encountered an unrecoverable error related to handling a request".into(),
PgPool(e) => format!("{}", e), PgPool(e) => format!("{}", e),
Pg(e) => format!("{}", e), Pg(e) => format!("{}", e),
}; };

View File

@ -11,7 +11,7 @@ use std::convert::TryFrom;
#[derive(Clone, Debug)] #[derive(Clone, Debug)]
pub struct PgPool { pub struct PgPool {
pub conn: r2d2::Pool<PostgresConnectionManager<postgres::NoTls>>, conn: r2d2::Pool<PostgresConnectionManager<postgres::NoTls>>,
whitelist_mode: bool, whitelist_mode: bool,
} }
@ -19,7 +19,7 @@ type Result<T> = std::result::Result<T, err::RequestErr>;
type Rejectable<T> = std::result::Result<T, warp::Rejection>; type Rejectable<T> = std::result::Result<T, warp::Rejection>;
impl PgPool { impl PgPool {
pub fn new(pg_cfg: &config::Postgres, whitelist_mode: bool) -> Result<Self> { pub(crate) fn new(pg_cfg: &config::Postgres, whitelist_mode: bool) -> Result<Self> {
let mut cfg = postgres::Config::new(); let mut cfg = postgres::Config::new();
cfg.user(&pg_cfg.user) cfg.user(&pg_cfg.user)
.host(&*pg_cfg.host.to_string()) .host(&*pg_cfg.host.to_string())
@ -40,7 +40,7 @@ impl PgPool {
}) })
} }
pub fn select_user(self, token: &Option<String>) -> Rejectable<UserData> { pub(crate) fn select_user(self, token: &Option<String>) -> Rejectable<UserData> {
let mut conn = self.conn.get().map_err(warp::reject::custom)?; let mut conn = self.conn.get().map_err(warp::reject::custom)?;
if let Some(token) = token { if let Some(token) = token {
@ -89,7 +89,7 @@ LIMIT 1",
} }
} }
pub fn select_hashtag_id(self, tag_name: &str) -> Rejectable<i64> { pub(crate) fn select_hashtag_id(self, tag_name: &str) -> Rejectable<i64> {
let mut conn = self.conn.get().map_err(warp::reject::custom)?; let mut conn = self.conn.get().map_err(warp::reject::custom)?;
conn.query("SELECT id FROM tags WHERE name = $1 LIMIT 1", &[&tag_name]) conn.query("SELECT id FROM tags WHERE name = $1 LIMIT 1", &[&tag_name])
.map_err(warp::reject::custom)? .map_err(warp::reject::custom)?
@ -102,7 +102,7 @@ LIMIT 1",
/// ///
/// **NOTE**: because we check this when the user connects, it will not include any blocks /// **NOTE**: because we check this when the user connects, it will not include any blocks
/// the user adds until they refresh/reconnect. /// the user adds until they refresh/reconnect.
pub fn select_blocked_users(self, user_id: Id) -> Rejectable<HashSet<Id>> { pub(crate) fn select_blocked_users(self, user_id: Id) -> Rejectable<HashSet<Id>> {
let mut conn = self.conn.get().map_err(warp::reject::custom)?; let mut conn = self.conn.get().map_err(warp::reject::custom)?;
conn.query( conn.query(
"SELECT target_account_id FROM blocks WHERE account_id = $1 "SELECT target_account_id FROM blocks WHERE account_id = $1
@ -114,11 +114,12 @@ LIMIT 1",
.map(|row| Ok(Id(row.get(0)))) .map(|row| Ok(Id(row.get(0))))
.collect() .collect()
} }
/// Query Postgres for everyone who has blocked the user /// Query Postgres for everyone who has blocked the user
/// ///
/// **NOTE**: because we check this when the user connects, it will not include any blocks /// **NOTE**: because we check this when the user connects, it will not include any blocks
/// the user adds until they refresh/reconnect. /// the user adds until they refresh/reconnect.
pub fn select_blocking_users(self, user_id: Id) -> Rejectable<HashSet<Id>> { pub(crate) fn select_blocking_users(self, user_id: Id) -> Rejectable<HashSet<Id>> {
let mut conn = self.conn.get().map_err(warp::reject::custom)?; let mut conn = self.conn.get().map_err(warp::reject::custom)?;
conn.query( conn.query(
"SELECT account_id FROM blocks WHERE target_account_id = $1", "SELECT account_id FROM blocks WHERE target_account_id = $1",
@ -134,7 +135,7 @@ LIMIT 1",
/// ///
/// **NOTE**: because we check this when the user connects, it will not include any blocks /// **NOTE**: because we check this when the user connects, it will not include any blocks
/// the user adds until they refresh/reconnect. /// the user adds until they refresh/reconnect.
pub fn select_blocked_domains(self, user_id: Id) -> Rejectable<HashSet<String>> { pub(crate) fn select_blocked_domains(self, user_id: Id) -> Rejectable<HashSet<String>> {
let mut conn = self.conn.get().map_err(warp::reject::custom)?; let mut conn = self.conn.get().map_err(warp::reject::custom)?;
conn.query( conn.query(
"SELECT domain FROM account_domain_blocks WHERE account_id = $1", "SELECT domain FROM account_domain_blocks WHERE account_id = $1",
@ -147,7 +148,7 @@ LIMIT 1",
} }
/// Test whether a user owns a list /// Test whether a user owns a list
pub fn user_owns_list(self, user_id: Id, list_id: i64) -> Rejectable<bool> { pub(crate) fn user_owns_list(self, user_id: Id, list_id: i64) -> Rejectable<bool> {
let mut conn = self.conn.get().map_err(warp::reject::custom)?; let mut conn = self.conn.get().map_err(warp::reject::custom)?;
// For the Postgres query, `id` = list number; `account_id` = user.id // For the Postgres query, `id` = list number; `account_id` = user.id
let rows = &conn let rows = &conn

View File

@ -4,16 +4,16 @@ use warp::filters::BoxedFilter;
use warp::Filter as WarpFilter; use warp::Filter as WarpFilter;
#[derive(Debug)] #[derive(Debug)]
pub struct Query { pub(crate) struct Query {
pub access_token: Option<String>, pub(crate) access_token: Option<String>,
pub stream: String, pub(crate) stream: String,
pub media: bool, pub(crate) media: bool,
pub hashtag: String, pub(crate) hashtag: String,
pub list: i64, pub(crate) list: i64,
} }
impl Query { impl Query {
pub fn update_access_token( pub(crate) fn update_access_token(
self, self,
token: Option<String>, token: Option<String>,
) -> Result<Self, warp::reject::Rejection> { ) -> Result<Self, warp::reject::Rejection> {
@ -30,17 +30,17 @@ impl Query {
macro_rules! make_query_type { macro_rules! make_query_type {
(Stream => $parameter:tt:$type:ty) => { (Stream => $parameter:tt:$type:ty) => {
#[derive(Deserialize, Debug, Default)] #[derive(Deserialize, Debug, Default)]
pub struct Stream { pub(crate) struct Stream {
pub $parameter: $type, pub(crate) $parameter: $type,
} }
}; };
($name:tt => $parameter:tt:$type:ty) => { ($name:tt => $parameter:tt:$type:ty) => {
#[derive(Deserialize, Debug, Default)] #[derive(Deserialize, Debug, Default)]
pub struct $name { pub(crate) struct $name {
pub $parameter: $type, pub(crate) $parameter: $type,
} }
impl $name { impl $name {
pub fn to_filter() -> BoxedFilter<(Self,)> { pub(crate) fn to_filter() -> BoxedFilter<(Self,)> {
warp::query() warp::query()
.or(warp::any().map(Self::default)) .or(warp::any().map(Self::default))
.unify() .unify()
@ -51,7 +51,7 @@ macro_rules! make_query_type {
} }
make_query_type!(Media => only_media:String); make_query_type!(Media => only_media:String);
impl Media { impl Media {
pub fn is_truthy(&self) -> bool { pub(crate) fn is_truthy(&self) -> bool {
self.only_media == "true" || self.only_media == "1" self.only_media == "true" || self.only_media == "1"
} }
} }
@ -65,19 +65,10 @@ impl ToString for Stream {
} }
} }
// pub fn optional_media_query() -> BoxedFilter<(Media,)> { pub(super) struct OptionalAccessToken;
// warp::query()
// .or(warp::any().map(|| Media {
// only_media: "false".to_owned(),
// }))
// .unify()
// .boxed()
// }
pub struct OptionalAccessToken;
impl OptionalAccessToken { impl OptionalAccessToken {
pub fn from_sse_header() -> warp::filters::BoxedFilter<(Option<String>,)> { pub(super) fn from_sse_header() -> warp::filters::BoxedFilter<(Option<String>,)> {
let from_header = warp::header::header::<String>("authorization").map(|auth: String| { let from_header = warp::header::header::<String>("authorization").map(|auth: String| {
match auth.split(' ').nth(1) { match auth.split(' ').nth(1) {
Some(s) => Some(s.to_string()), Some(s) => Some(s.to_string()),
@ -88,7 +79,7 @@ impl OptionalAccessToken {
from_header.or(no_token).unify().boxed() from_header.or(no_token).unify().boxed()
} }
pub fn from_ws_header() -> warp::filters::BoxedFilter<(Option<String>,)> { pub(super) fn from_ws_header() -> warp::filters::BoxedFilter<(Option<String>,)> {
let from_header = warp::header::header::<String>("Sec-Websocket-Protocol").map(Some); let from_header = warp::header::header::<String>("Sec-Websocket-Protocol").map(Some);
let no_token = warp::any().map(|| None); let no_token = warp::any().map(|| None);

View File

@ -17,17 +17,17 @@ use warp::reject::Rejection;
#[derive(Clone, Debug, PartialEq)] #[derive(Clone, Debug, PartialEq)]
pub struct Subscription { pub struct Subscription {
pub timeline: Timeline, pub timeline: Timeline,
pub allowed_langs: HashSet<String>, pub(crate) allowed_langs: HashSet<String>,
pub blocks: Blocks, pub(crate) blocks: Blocks,
pub hashtag_name: Option<String>, pub(crate) hashtag_name: Option<String>,
pub access_token: Option<String>, pub access_token: Option<String>,
} }
#[derive(Clone, Default, Debug, PartialEq)] #[derive(Clone, Default, Debug, PartialEq)]
pub struct Blocks { pub(crate) struct Blocks {
pub blocked_domains: HashSet<String>, pub(crate) blocked_domains: HashSet<String>,
pub blocked_users: HashSet<Id>, pub(crate) blocked_users: HashSet<Id>,
pub blocking_users: HashSet<Id>, pub(crate) blocking_users: HashSet<Id>,
} }
impl Default for Subscription { impl Default for Subscription {

View File

@ -1,5 +1,5 @@
pub use self::err::TimelineErr; pub(crate) use self::err::TimelineErr;
pub use self::inner::{Content, Reach, Scope, Stream, UserData}; pub(crate) use self::inner::{Content, Reach, Scope, Stream, UserData};
use super::query::Query; use super::query::Query;
use lru::LruCache; use lru::LruCache;
@ -11,14 +11,14 @@ mod inner;
type Result<T> = std::result::Result<T, TimelineErr>; type Result<T> = std::result::Result<T, TimelineErr>;
#[derive(Clone, Debug, Copy, Eq, Hash, PartialEq)] #[derive(Clone, Debug, Copy, Eq, Hash, PartialEq)]
pub struct Timeline(pub Stream, pub Reach, pub Content); pub struct Timeline(pub(crate) Stream, pub(crate) Reach, pub(crate) Content);
impl Timeline { impl Timeline {
pub fn empty() -> Self { pub fn empty() -> Self {
Self(Stream::Unset, Reach::Local, Content::Notification) Self(Stream::Unset, Reach::Local, Content::Notification)
} }
pub fn to_redis_raw_timeline(&self, hashtag: Option<&String>) -> Result<String> { pub(crate) fn to_redis_raw_timeline(&self, hashtag: Option<&String>) -> Result<String> {
// TODO -- does this need to account for namespaces? // TODO -- does this need to account for namespaces?
use {Content::*, Reach::*, Stream::*, TimelineErr::*}; use {Content::*, Reach::*, Stream::*, TimelineErr::*};
@ -46,7 +46,10 @@ impl Timeline {
}) })
} }
pub fn from_redis_text(timeline: &str, cache: &mut LruCache<String, i64>) -> Result<Self> { pub(crate) fn from_redis_text(
timeline: &str,
cache: &mut LruCache<String, i64>,
) -> Result<Self> {
use {Content::*, Reach::*, Stream::*, TimelineErr::*}; use {Content::*, Reach::*, Stream::*, TimelineErr::*};
let mut tag_id = |t: &str| cache.get(&t.to_string()).map_or(Err(BadTag), |id| Ok(*id)); let mut tag_id = |t: &str| cache.get(&t.to_string()).map_or(Err(BadTag), |id| Ok(*id));
@ -65,7 +68,10 @@ impl Timeline {
}) })
} }
pub fn from_query_and_user(q: &Query, user: &UserData) -> std::result::Result<Self, Rejection> { pub(crate) fn from_query_and_user(
q: &Query,
user: &UserData,
) -> std::result::Result<Self, Rejection> {
use {warp::reject::custom, Content::*, Reach::*, Scope::*, Stream::*}; use {warp::reject::custom, Content::*, Reach::*, Scope::*, Stream::*};
Ok(match q.stream.as_ref() { Ok(match q.stream.as_ref() {

View File

@ -5,7 +5,7 @@ use hashbrown::HashSet;
use std::convert::TryFrom; use std::convert::TryFrom;
#[derive(Clone, Debug, Copy, Eq, Hash, PartialEq)] #[derive(Clone, Debug, Copy, Eq, Hash, PartialEq)]
pub enum Stream { pub(crate) enum Stream {
User(Id), User(Id),
List(i64), List(i64),
Direct(i64), Direct(i64),
@ -15,20 +15,20 @@ pub enum Stream {
} }
#[derive(Clone, Debug, Copy, Eq, Hash, PartialEq)] #[derive(Clone, Debug, Copy, Eq, Hash, PartialEq)]
pub enum Reach { pub(crate) enum Reach {
Local, Local,
Federated, Federated,
} }
#[derive(Clone, Debug, Copy, Eq, Hash, PartialEq)] #[derive(Clone, Debug, Copy, Eq, Hash, PartialEq)]
pub enum Content { pub(crate) enum Content {
All, All,
Media, Media,
Notification, Notification,
} }
#[derive(Clone, Debug, PartialEq, Eq, Hash)] #[derive(Clone, Debug, PartialEq, Eq, Hash)]
pub enum Scope { pub(crate) enum Scope {
Read, Read,
Statuses, Statuses,
Notifications, Notifications,
@ -53,10 +53,10 @@ impl TryFrom<&str> for Scope {
} }
} }
pub struct UserData { pub(crate) struct UserData {
pub id: Id, pub(crate) id: Id,
pub allowed_langs: HashSet<String>, pub(crate) allowed_langs: HashSet<String>,
pub scopes: HashSet<Scope>, pub(crate) scopes: HashSet<Scope>,
} }
impl UserData { impl UserData {

View File

@ -3,7 +3,7 @@
pub mod redis; pub mod redis;
pub mod stream; pub mod stream;
pub use redis::{Manager, ManagerErr}; pub(crate) use redis::ManagerErr;
#[cfg(feature = "bench")] #[cfg(feature = "bench")]
pub use redis::msg::{RedisMsg, RedisParseOutput}; pub use redis::msg::{RedisMsg, RedisParseOutput};

View File

@ -1,18 +1,19 @@
pub mod connection; mod connection;
mod manager; mod manager;
pub mod msg; mod msg;
pub use connection::{RedisConn, RedisConnErr}; pub(crate) use connection::{RedisConn, RedisConnErr};
pub use manager::{Manager, ManagerErr}; pub use manager::Manager;
pub use msg::RedisParseErr; pub(crate) use manager::ManagerErr;
pub(crate) use msg::RedisParseErr;
pub enum RedisCmd { pub(crate) enum RedisCmd {
Subscribe, Subscribe,
Unsubscribe, Unsubscribe,
} }
impl RedisCmd { impl RedisCmd {
pub fn into_sendable(self, tl: &str) -> (Vec<u8>, Vec<u8>) { pub(crate) fn into_sendable(self, tl: &str) -> (Vec<u8>, Vec<u8>) {
match self { match self {
RedisCmd::Subscribe => ( RedisCmd::Subscribe => (
[ [

View File

@ -1,5 +1,5 @@
mod err; mod err;
pub use err::RedisConnErr; pub(crate) use err::RedisConnErr;
use super::msg::{RedisParseErr, RedisParseOutput}; use super::msg::{RedisParseErr, RedisParseOutput};
use super::{ManagerErr, RedisCmd}; use super::{ManagerErr, RedisCmd};
@ -18,7 +18,7 @@ use std::time::Duration;
type Result<T> = std::result::Result<T, RedisConnErr>; type Result<T> = std::result::Result<T, RedisConnErr>;
#[derive(Debug)] #[derive(Debug)]
pub struct RedisConn { pub(crate) struct RedisConn {
primary: TcpStream, primary: TcpStream,
secondary: TcpStream, secondary: TcpStream,
redis_namespace: Option<String>, redis_namespace: Option<String>,
@ -29,7 +29,7 @@ pub struct RedisConn {
} }
impl RedisConn { impl RedisConn {
pub fn new(redis_cfg: &Redis) -> Result<Self> { pub(crate) fn new(redis_cfg: &Redis) -> Result<Self> {
let addr = [&*redis_cfg.host, ":", &*redis_cfg.port.to_string()].concat(); let addr = [&*redis_cfg.host, ":", &*redis_cfg.port.to_string()].concat();
let conn = Self::new_connection(&addr, redis_cfg.password.as_ref())?; let conn = Self::new_connection(&addr, redis_cfg.password.as_ref())?;
@ -50,7 +50,7 @@ impl RedisConn {
Ok(redis_conn) Ok(redis_conn)
} }
pub fn poll_redis(&mut self) -> Poll<Option<(Timeline, Event)>, ManagerErr> { pub(crate) fn poll_redis(&mut self) -> Poll<Option<(Timeline, Event)>, ManagerErr> {
loop { loop {
match self.primary.read(&mut self.redis_input[self.cursor..]) { match self.primary.read(&mut self.redis_input[self.cursor..]) {
Ok(n) => { Ok(n) => {
@ -108,26 +108,15 @@ impl RedisConn {
self.redis_input[acc] = cur.expect("TODO"); self.redis_input[acc] = cur.expect("TODO");
acc + 1 acc + 1
}); });
// self.cursor = 0;
// for (i, byte) in [leftover.as_bytes(), invalid_bytes]
// .concat()
// .bytes()
// .enumerate()
// {
// self.redis_input[i] = byte.expect("TODO");
// self.cursor += 1;
// }
res res
} }
pub fn update_cache(&mut self, hashtag: String, id: i64) { pub(crate) fn update_cache(&mut self, hashtag: String, id: i64) {
self.tag_id_cache.put(hashtag.clone(), id); self.tag_id_cache.put(hashtag.clone(), id);
self.tag_name_cache.put(id, hashtag); self.tag_name_cache.put(id, hashtag);
} }
pub fn send_cmd(&mut self, cmd: RedisCmd, timeline: &Timeline) -> Result<()> { pub(crate) fn send_cmd(&mut self, cmd: RedisCmd, timeline: &Timeline) -> Result<()> {
let hashtag = match timeline { let hashtag = match timeline {
Timeline(Stream::Hashtag(id), _, _) => self.tag_name_cache.get(id), Timeline(Stream::Hashtag(id), _, _) => self.tag_name_cache.get(id),
_non_hashtag_timeline => None, _non_hashtag_timeline => None,

View File

@ -13,7 +13,7 @@ pub enum RedisConnErr {
} }
impl RedisConnErr { impl RedisConnErr {
pub fn with_addr<T: AsRef<str>>(address: T, inner: std::io::Error) -> Self { pub(crate) fn with_addr<T: AsRef<str>>(address: T, inner: std::io::Error) -> Self {
Self::ConnectionErr { Self::ConnectionErr {
addr: address.as_ref().to_string(), addr: address.as_ref().to_string(),
inner, inner,

View File

@ -2,7 +2,7 @@
//! polled by the correct `ClientAgent`. Also manages sububscriptions and //! polled by the correct `ClientAgent`. Also manages sububscriptions and
//! unsubscriptions to/from Redis. //! unsubscriptions to/from Redis.
mod err; mod err;
pub use err::ManagerErr; pub(crate) use err::ManagerErr;
use super::{RedisCmd, RedisConn}; use super::{RedisCmd, RedisConn};
use crate::config; use crate::config;
@ -68,7 +68,7 @@ impl Manager {
}; };
} }
pub fn unsubscribe(&mut self, tl: Timeline) -> Result<()> { pub(crate) fn unsubscribe(&mut self, tl: Timeline) -> Result<()> {
let number_of_subscriptions = self let number_of_subscriptions = self
.clients_per_timeline .clients_per_timeline
.entry(tl) .entry(tl)

View File

@ -19,28 +19,24 @@
//! Read that as: an array with three elements: the first element is a bulk string with //! Read that as: an array with three elements: the first element is a bulk string with
//! three characters, the second is a bulk string with ten characters, and the third is a //! three characters, the second is a bulk string with ten characters, and the third is a
//! bulk string with 1,386 characters. //! bulk string with 1,386 characters.
use self::RedisParseOutput::*;
pub use err::RedisParseErr;
use std::convert::{TryFrom, TryInto};
use std::str;
mod err; mod err;
pub use err::RedisParseErr;
use self::RedisParseOutput::*;
use std::{
convert::{TryFrom, TryInto},
str,
};
#[derive(Debug, Clone, PartialEq)] #[derive(Debug, Clone, PartialEq)]
pub enum RedisParseOutput<'a> { pub(crate) enum RedisParseOutput<'a> {
Msg(RedisMsg<'a>), Msg(RedisMsg<'a>),
NonMsg(&'a str), NonMsg(&'a str),
} }
#[derive(Debug, Clone, PartialEq)] #[derive(Debug, Clone, PartialEq)]
pub struct RedisMsg<'a> { pub(crate) struct RedisMsg<'a> {
pub timeline_txt: &'a str, pub(crate) timeline_txt: &'a str,
pub event_txt: &'a str, pub(crate) event_txt: &'a str,
pub leftover_input: &'a str, pub(crate) leftover_input: &'a str,
} }
impl<'a> TryFrom<&'a str> for RedisParseOutput<'a> { impl<'a> TryFrom<&'a str> for RedisParseOutput<'a> {