Renamed service to api

api/src/airports/mod.rs

@@ -0,0 +1,5 @@
+mod model;
+mod routes;
+
+pub use model::*;
+pub use routes::init_routes;

api/src/airports/model.rs

@@ -0,0 +1,432 @@
+use std::fmt::Display;
+use std::str::FromStr;
+
+use crate::db;
+use crate::error::{ApiError, ApiResult};
+use crate::db::schema::airports;
+use diesel::prelude::*;
+use diesel::sql_query;
+use log::error;
+use postgis_diesel::types::*;
+use serde::{Deserialize, Serialize};
+
+#[derive(Serialize, Deserialize)]
+pub struct Runway {
+  pub id: String,
+  pub length_ft: f32,
+  pub width_ft: f32,
+  pub surface: String,
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct Frequency {
+  pub id: String,
+  pub frequency_mhz: f32,
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct Airport {
+  pub icao: String,
+  #[serde(skip_serializing_if = "Option::is_none")]
+  pub iata: Option<String>,
+  #[serde(skip_serializing_if = "Option::is_none")]
+  pub local: Option<String>,
+  pub name: String,
+  pub category: AirportCategory,
+  pub iso_country: String,
+  pub iso_region: String,
+  pub municipality: String,
+  pub elevation_ft: f32,
+  pub latitude: f64,
+  pub longitude: f64,
+  #[serde(skip_serializing_if = "Option::is_none")]
+  pub has_tower: Option<bool>,
+  #[serde(skip_serializing_if = "Option::is_none")]
+  pub has_beacon: Option<bool>,
+  pub runways: Vec<Runway>,
+  pub frequencies: Vec<Frequency>,
+  pub public: bool,
+}
+
+#[derive(Serialize, Deserialize, Clone, Debug)]
+pub struct AirportMetarCache {
+  pub icao: String,
+  pub has_metar: bool,
+  pub last_checked: chrono::NaiveDateTime,
+}
+
+impl Into<QueryAirport> for Airport {
+  fn into(self) -> QueryAirport {
+    return QueryAirport {
+      icao: self.icao.clone(),
+      category: self.category.clone().to_string(),
+      name: self.name.clone(),
+      elevation_ft: self.elevation_ft,
+      iso_country: self.iso_country.clone(),
+      iso_region: self.iso_region.clone(),
+      municipality: self.municipality.clone(),
+      has_metar: false,
+      point: Point::new(self.longitude, self.latitude, Some(4326)),
+      data: match serde_json::to_value(&self) {
+        Ok(d) => d,
+        Err(err) => {
+          error!("{}", err);
+          serde_json::Value::Null
+        }
+      },
+    };
+  }
+}
+
+impl From<QueryAirport> for Airport {
+  fn from(airport: QueryAirport) -> Self {
+    serde_json::from_value(airport.data).unwrap()
+  }
+}
+
+#[derive(Serialize, Deserialize, Clone, Debug)]
+pub enum AirportCategory {
+  #[serde(rename = "small_airport")]
+  Small,
+  #[serde(rename = "medium_airport")]
+  Medium,
+  #[serde(rename = "large_airport")]
+  Large,
+  #[serde(rename = "heliport")]
+  Heliport,
+  #[serde(rename = "closed")]
+  Closed,
+  #[serde(rename = "seaplane_base")]
+  Seaplane,
+  #[serde(rename = "balloonport")]
+  Balloonport,
+  #[serde(rename = "unknown")]
+  Unknown,
+}
+
+impl FromStr for AirportCategory {
+  type Err = ();
+  fn from_str(s: &str) -> Result<Self, Self::Err> {
+    match s {
+      "small_airport" => Ok(AirportCategory::Small),
+      "medium_airport" => Ok(AirportCategory::Medium),
+      "large_airport" => Ok(AirportCategory::Large),
+      "heliport" => Ok(AirportCategory::Heliport),
+      "closed" => Ok(AirportCategory::Closed),
+      "seaplane_base" => Ok(AirportCategory::Seaplane),
+      "balloonport" => Ok(AirportCategory::Balloonport),
+      _ => Ok(AirportCategory::Unknown),
+    }
+  }
+}
+
+impl Display for AirportCategory {
+  fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+    match self {
+      AirportCategory::Small => write!(f, "small_airport"),
+      AirportCategory::Medium => write!(f, "medium_airport"),
+      AirportCategory::Large => write!(f, "large_airport"),
+      AirportCategory::Heliport => write!(f, "heliport"),
+      AirportCategory::Closed => write!(f, "closed"),
+      AirportCategory::Seaplane => write!(f, "seaplane_base"),
+      AirportCategory::Balloonport => write!(f, "balloonport"),
+      AirportCategory::Unknown => write!(f, "unknown"),
+    }
+  }
+}
+
+#[derive(Serialize, Deserialize, AsChangeset, Insertable, Queryable, QueryableByName)]
+#[diesel(table_name = airports)]
+pub struct QueryAirport {
+  pub icao: String,
+  pub category: String,
+  pub name: String,
+  pub elevation_ft: f32,
+  pub iso_country: String,
+  pub iso_region: String,
+  pub municipality: String,
+  pub has_metar: bool,
+  pub point: Point,
+  pub data: serde_json::Value,
+}
+
+#[derive(Debug)]
+pub struct QueryFilters {
+  pub icaos: Option<Vec<String>>,
+  pub name: Option<String>,
+  pub bounds: Option<Polygon<Point>>,
+  pub categories: Option<Vec<AirportCategory>>,
+  pub order_field: Option<QueryOrderField>,
+  pub order_by: Option<QueryOrderBy>,
+  pub has_metar: Option<bool>,
+}
+
+impl Default for QueryFilters {
+  fn default() -> Self {
+    QueryFilters {
+      icaos: None,
+      name: None,
+      bounds: None,
+      categories: None,
+      order_field: None,
+      order_by: None,
+      has_metar: None,
+    }
+  }
+}
+
+#[derive(Debug)]
+pub enum QueryOrderBy {
+  Asc,
+  Desc,
+}
+
+impl FromStr for QueryOrderBy {
+  type Err = ();
+  fn from_str(s: &str) -> Result<Self, Self::Err> {
+    match s {
+      "asc" => Ok(QueryOrderBy::Asc),
+      "desc" => Ok(QueryOrderBy::Desc),
+      _ => Err(()),
+    }
+  }
+}
+
+#[derive(Debug)]
+pub enum QueryOrderField {
+  Icao,
+  Name,
+  Category,
+  Country,
+  Region,
+  Municipality,
+}
+
+impl FromStr for QueryOrderField {
+  type Err = ();
+  fn from_str(s: &str) -> Result<Self, Self::Err> {
+    match s {
+      "icao" => Ok(QueryOrderField::Icao),
+      "name" => Ok(QueryOrderField::Name),
+      "category" => Ok(QueryOrderField::Category),
+      "iso_country" => Ok(QueryOrderField::Country),
+      "iso_region" => Ok(QueryOrderField::Region),
+      "municipality" => Ok(QueryOrderField::Municipality),
+      _ => Err(()),
+    }
+  }
+}
+
+impl QueryAirport {
+  pub fn get_all(filters: &QueryFilters, limit: i32, page: i32) -> ApiResult<Vec<Self>> {
+    let mut conn = db::connection()?;
+    let mut query: String = "SELECT * FROM airports".to_string();
+    query = format!("{} {}", query, QueryAirport::build_filter_query(&filters)?);
+
+    query = format!("{} ORDER BY has_metar DESC", query);
+    if let Some(order_by) = &filters.order_by {
+      match order_by {
+        QueryOrderBy::Asc => {
+          if let Some(order_field) = &filters.order_field {
+            query = match order_field {
+              QueryOrderField::Icao => format!("{}, icao ASC", query),
+              QueryOrderField::Name => format!("{}, name ASC", query),
+              QueryOrderField::Category => format!("{}, category ASC", query),
+              QueryOrderField::Country => format!("{}, iso_country ASC", query),
+              QueryOrderField::Region => format!("{}, iso_region ASC", query),
+              QueryOrderField::Municipality => format!("{}, municipality ASC", query),
+            };
+          };
+        }
+        QueryOrderBy::Desc => {
+          if let Some(order_field) = &filters.order_field {
+            query = match order_field {
+              QueryOrderField::Icao => format!("{}, icao DESC", query),
+              QueryOrderField::Name => format!("{}, name DESC", query),
+              QueryOrderField::Category => format!("{}, category DESC", query),
+              QueryOrderField::Country => format!("{}, iso_country DESC", query),
+              QueryOrderField::Region => format!("{}, iso_region DESC", query),
+              QueryOrderField::Municipality => format!("{}, municipality DESC", query),
+            };
+          };
+        }
+      }
+    }
+    // Limit query to page and limit
+    query = format!("{} LIMIT {} OFFSET {}", query, limit, (page - 1) * limit);
+
+    let airports: Vec<QueryAirport> = match sql_query(query).load(&mut conn) {
+      Ok(a) => a,
+      Err(err) => {
+        return Err(ApiError {
+          status: 500,
+          message: format!("{}", err),
+        })
+      }
+    };
+    Ok(airports)
+  }
+
+  pub fn get_count(filters: &QueryFilters) -> ApiResult<i64> {
+    let mut conn = db::connection()?;
+    let mut query = "SELECT COUNT(*) FROM airports".to_string();
+    query = format!("{} {}", query, QueryAirport::build_filter_query(&filters)?);
+
+    // TODO: Fix this to use get_result() instead of building this table to do the load()
+    diesel::table! {
+      airports (count) {
+        count -> BigInt,
+      }
+    }
+    #[derive(Debug, Queryable, QueryableByName)]
+    #[diesel(table_name = airports)]
+    struct Count {
+      count: i64,
+    }
+
+    let count: Vec<Count> = match sql_query(query).load(&mut conn) {
+      Ok(a) => a,
+      Err(err) => {
+        return Err(ApiError {
+          status: 500,
+          message: format!("{}", err),
+        })
+      }
+    };
+    return Ok(count[0].count);
+  }
+
+  // TODO: Unsafe query, need to sanitize inputs
+  fn build_filter_query(filters: &QueryFilters) -> ApiResult<String> {
+    let mut query = "".to_string();
+    let mut parts: Vec<String> = vec![];
+
+    if let Some(bounds) = &filters.bounds {
+      // convert bounds to a WKT polygon
+      if bounds.rings.len() > 1 {
+        return Err(ApiError {
+          status: 400,
+          message: "Only one polygon is allowed".to_string(),
+        });
+      } else {
+        let mut points: Vec<String> = vec![];
+        bounds.rings.iter().for_each(|ring| {
+          ring.iter().for_each(|point| {
+            points.push(format!("{} {}", point.get_x(), point.get_y()));
+          });
+        });
+        let bounds = format!("POLYGON(({}))", points.join(","));
+        parts.push(format!(
+          "ST_Contains(ST_GeomFromText('{}', 4326), point)",
+          bounds
+        ));
+      }
+    }
+    if let Some(categories) = &filters.categories {
+      parts.push(format!(
+        "({})",
+        categories
+          .iter()
+          .map(|category| format!("category = '{}'", category.to_string()))
+          .collect::<Vec<String>>()
+          .join(" OR ")
+      ));
+    }
+    fn sanitize_icao(icao: &str) -> String {
+      // Sanitize search to only allow [a-zA-Z0-9-\\s]
+      icao
+        .chars()
+        .filter(|c| c.is_alphanumeric() || *c == '-' || *c == ' ')
+        .collect::<String>()
+    }
+    if &filters.icaos.is_some() == &true && &filters.name.is_some() == &true {
+      let icaos = filters.icaos.as_ref().unwrap();
+      let name = sanitize_icao(filters.name.as_ref().unwrap());
+      let icao_part = format!(
+        "({})",
+        icaos
+          .iter()
+          .map(|icao| format!("icao ILIKE '{}'", sanitize_icao(icao)))
+          .collect::<Vec<String>>()
+          .join(" OR ")
+      );
+      let name_part = format!("name ILIKE '%{}%'", name);
+      parts.push(format!("({} OR {})", icao_part, name_part));
+    } else if let Some(icaos) = &filters.icaos {
+      parts.push(format!(
+        "({})",
+        icaos
+          .iter()
+          .map(|icao| format!("icao ILIKE '{}'", sanitize_icao(icao)))
+          .collect::<Vec<String>>()
+          .join(" OR ")
+      ));
+    } else if let Some(name) = &filters.name {
+      let search = sanitize_icao(name);
+      parts.push(format!("name ILIKE '%{}%'", search));
+    }
+    if let Some(has_metar) = &filters.has_metar {
+      parts.push(format!("has_metar = {}", has_metar));
+    }
+
+    if parts.len() > 0 {
+      query = format!("{} WHERE {}", query, parts.join(" AND "));
+    }
+
+    return Ok(query);
+  }
+
+  pub fn get(icao: &str) -> ApiResult<Self> {
+    let mut conn = db::connection()?;
+    let airport = airports::table
+      .filter(airports::icao.eq(icao))
+      .first(&mut conn)?;
+    Ok(airport)
+  }
+
+  pub fn insert(airport: Self) -> ApiResult<Self> {
+    let mut conn: r2d2::PooledConnection<diesel::r2d2::ConnectionManager<PgConnection>> =
+      db::connection()?;
+    let airport = Self::from(airport);
+    let airport = diesel::insert_into(airports::table)
+      .values(airport)
+      .on_conflict_do_nothing()
+      .get_result(&mut conn)?;
+    Ok(airport)
+  }
+
+  pub fn insert_all(airports: Vec<Self>) -> ApiResult<Vec<Self>> {
+    let mut conn: r2d2::PooledConnection<diesel::r2d2::ConnectionManager<PgConnection>> =
+      db::connection()?;
+    let mut inserted_airports: Vec<Self> = vec![];
+    for airport in airports {
+      let airport = Self::from(airport);
+      let airport = diesel::insert_into(airports::table)
+        .values(airport)
+        .on_conflict_do_nothing()
+        .get_result(&mut conn)?;
+      inserted_airports.push(airport);
+    }
+    Ok(inserted_airports)
+  }
+
+  pub fn update(airport: Self) -> ApiResult<Self> {
+    let mut conn = db::connection()?;
+    let airport = diesel::update(airports::table)
+      .filter(airports::icao.eq(airport.icao.clone()))
+      .set(airport)
+      .get_result(&mut conn)?;
+    Ok(airport)
+  }
+
+  pub fn delete(icao: Option<String>) -> ApiResult<usize> {
+    let mut conn = db::connection()?;
+    let res = match icao {
+      Some(icao) => {
+        diesel::delete(airports::table.filter(airports::icao.eq(icao))).execute(&mut conn)?
+      }
+      None => diesel::delete(airports::table).execute(&mut conn)?,
+    };
+    Ok(res)
+  }
+}

api/src/airports/routes.rs

@@ -0,0 +1,303 @@
+use std::str::FromStr;
+use futures_util::stream::StreamExt as _;
+
+use crate::{
+  airports::{QueryAirport, QueryFilters, QueryOrderField, QueryOrderBy, Airport, AirportCategory},
+  db::{Response, Metadata},
+  auth::{Auth, verify_role},
+};
+use actix_multipart::Multipart;
+use actix_web::{delete, get, post, put, web, HttpResponse, HttpRequest, ResponseError};
+use log::{error, warn};
+use postgis_diesel::types::{Polygon, Point};
+use serde::{Serialize, Deserialize};
+
+#[derive(Debug, Serialize, Deserialize)]
+struct AirportsQuery {
+  icaos: Option<String>,
+  name: Option<String>,
+  bounds: Option<String>,
+  categories: Option<String>,
+  order_field: Option<String>,
+  order_by: Option<String>,
+  has_metar: Option<String>,
+  limit: Option<i32>,
+  page: Option<i32>,
+}
+
+#[post("/import")]
+async fn import_airports(mut payload: Multipart, auth: Auth) -> HttpResponse {
+  if let Err(err) = verify_role(&auth, "admin") {
+    return ResponseError::error_response(&err);
+  };
+
+  while let Some(item) = payload.next().await {
+    let mut bytes = web::BytesMut::new();
+    let mut field = match item {
+      Ok(field) => field,
+      Err(err) => return ResponseError::error_response(&err),
+    };
+
+    // Build bytes from chunks
+    while let Some(chunk) = field.next().await {
+      let data = match chunk {
+        Ok(data) => data,
+        Err(err) => {
+          error!("Failed to get chunk: {}", err);
+          return ResponseError::error_response(&err);
+        }
+      };
+      bytes.extend_from_slice(&data);
+    }
+
+    // Convert bytes to Vec<Airport>
+    let airports: Vec<Airport> = match serde_json::from_slice(&bytes) {
+      Ok(a) => a,
+      Err(err) => {
+        error!("Failed to parse JSON: {}", err);
+        return ResponseError::error_response(&err);
+      }
+    };
+
+    // Convert Vec<Airport> to Vec<QueryAirport> and insert into database
+    let query_airports: Vec<QueryAirport> = airports.into_iter().map(|a| a.into()).collect();
+    match QueryAirport::insert_all(query_airports) {
+      Ok(_) => {}
+      Err(err) => return ResponseError::error_response(&err),
+    };
+  }
+  HttpResponse::Ok().finish()
+}
+
+#[get("")]
+async fn get_airports(req: HttpRequest) -> HttpResponse {
+  let params = web::Query::<AirportsQuery>::from_query(req.query_string()).unwrap();
+  let mut filters = QueryFilters::default();
+  filters.icaos = match &params.icaos {
+    Some(i) => Some(i.split(",").map(|s| s.to_string()).collect()),
+    None => None,
+  };
+  filters.name = params.name.clone();
+  filters.categories = match &params.categories {
+    Some(c) => Some(
+      c.split(",")
+        .map(|s| AirportCategory::from_str(s).unwrap())
+        .collect(),
+    ),
+    None => None,
+  };
+  filters.bounds = match &params.bounds {
+    Some(b) => {
+      let bounds: Vec<&str> = b.split(",").collect();
+      if bounds.len() != 4 {
+        warn!("Expected 4 bounds, received {}: {}", bounds.len(), b);
+        return HttpResponse::UnprocessableEntity().body(format!(
+          "Received {}; expected NE_LAT,NE_LON,SW_LAT,SW_LON",
+          b
+        ));
+      }
+      let ne_lat = match bounds[0].parse::<f64>() {
+        Ok(b) => b,
+        Err(err) => {
+          warn!("{}", err);
+          return HttpResponse::UnprocessableEntity().body(format!("{}", err));
+        }
+      };
+      let ne_lon = match bounds[1].parse::<f64>() {
+        Ok(b) => b,
+        Err(err) => {
+          warn!("{}", err);
+          return HttpResponse::UnprocessableEntity().body(format!("{}", err));
+        }
+      };
+      let sw_lat = match bounds[2].parse::<f64>() {
+        Ok(b) => b,
+        Err(err) => {
+          warn!("{}", err);
+          return HttpResponse::UnprocessableEntity().body(format!("{}", err));
+        }
+      };
+      let sw_lon = match bounds[3].parse::<f64>() {
+        Ok(b) => b,
+        Err(err) => {
+          warn!("{}", err);
+          return HttpResponse::UnprocessableEntity().body(format!("{}", err));
+        }
+      };
+      let mut polygon: Polygon<Point> = Polygon::new(Some(4326));
+      polygon.add_point(Point {
+        x: sw_lon,
+        y: sw_lat,
+        srid: Some(4326),
+      });
+      polygon.add_point(Point {
+        x: ne_lon,
+        y: sw_lat,
+        srid: Some(4326),
+      });
+      polygon.add_point(Point {
+        x: ne_lon,
+        y: ne_lat,
+        srid: Some(4326),
+      });
+      polygon.add_point(Point {
+        x: sw_lon,
+        y: ne_lat,
+        srid: Some(4326),
+      });
+      polygon.add_point(Point {
+        x: sw_lon,
+        y: sw_lat,
+        srid: Some(4326),
+      });
+      Some(polygon)
+    }
+    None => None,
+  };
+
+  filters.order_by = match &params.order_by {
+    Some(o) => Some(QueryOrderBy::from_str(&o).unwrap()),
+    None => None,
+  };
+  filters.order_field = match &params.order_field {
+    Some(o) => Some(QueryOrderField::from_str(&o).unwrap()),
+    None => None,
+  };
+  filters.has_metar = match &params.has_metar {
+    Some(h) => Some(h.parse::<bool>().unwrap()),
+    None => None,
+  };
+
+  let limit = match params.limit {
+    Some(l) => l,
+    None => 100,
+  };
+  let page = match params.page {
+    Some(p) => p,
+    None => 1,
+  };
+  let total = match QueryAirport::get_count(&filters) {
+    Ok(t) => t,
+    Err(_) => 0,
+  };
+
+  match web::block(move || QueryAirport::get_all(&filters, limit, page))
+    .await
+    .unwrap()
+  {
+    Ok(a) => {
+      // Convert Vec<QueryAirport> to Vec<Airport>
+      let mut airports: Vec<Airport> = vec![];
+      for airport in a {
+        airports.push(airport.into());
+      }
+      HttpResponse::Ok().json(Response {
+        data: airports,
+        meta: Some(Metadata { page, limit, total }),
+      })
+    }
+    Err(err) => {
+      error!("{}", err);
+      err.to_http_response()
+    }
+  }
+}
+
+#[get("/{icao}")]
+async fn get_airport(icao: web::Path<String>) -> HttpResponse {
+  match QueryAirport::get(&icao.into_inner()) {
+    Ok(a) => {
+      let airport: Airport = a.into();
+      HttpResponse::Ok().json(airport)
+    }
+    Err(err) => {
+      error!("{}", err);
+      err.to_http_response()
+    }
+  }
+}
+
+#[post("")]
+async fn create_airport(airport: web::Json<Airport>, auth: Auth) -> HttpResponse {
+  let _ = match verify_role(&auth, "admin") {
+    Ok(_) => {}
+    Err(err) => return ResponseError::error_response(&err),
+  };
+  let query_airport: QueryAirport = airport.into_inner().into();
+  match QueryAirport::insert(query_airport) {
+    Ok(a) => {
+      let airport: Airport = a.into();
+      HttpResponse::Ok().json(airport)
+    }
+    Err(err) => {
+      error!("{}", err);
+      err.to_http_response()
+    }
+  }
+}
+
+#[put("/{icao}")]
+async fn update_airport(
+  _icao: web::Path<String>,
+  airport: web::Json<Airport>,
+  auth: Auth,
+) -> HttpResponse {
+  let _ = match verify_role(&auth, "admin") {
+    Ok(_) => {}
+    Err(err) => return ResponseError::error_response(&err),
+  };
+  let query_airport: QueryAirport = airport.into_inner().into();
+  match QueryAirport::update(query_airport) {
+    Ok(a) => {
+      let airport: Airport = a.into();
+      HttpResponse::Ok().json(airport)
+    }
+    Err(err) => {
+      error!("{}", err);
+      err.to_http_response()
+    }
+  }
+}
+
+#[delete("")]
+async fn delete_airports(auth: Auth) -> HttpResponse {
+  let _ = match verify_role(&auth, "admin") {
+    Ok(_) => {}
+    Err(err) => return ResponseError::error_response(&err),
+  };
+  match QueryAirport::delete(None) {
+    Ok(_) => HttpResponse::NoContent().finish(),
+    Err(err) => {
+      error!("{}", err);
+      err.to_http_response()
+    }
+  }
+}
+
+#[delete("/{icao}")]
+async fn delete_airport(icao: web::Path<String>, auth: Auth) -> HttpResponse {
+  let _ = match verify_role(&auth, "admin") {
+    Ok(_) => {}
+    Err(err) => return ResponseError::error_response(&err),
+  };
+  match QueryAirport::delete(Some(icao.into_inner())) {
+    Ok(_) => HttpResponse::NoContent().finish(),
+    Err(err) => {
+      error!("{}", err);
+      err.to_http_response()
+    }
+  }
+}
+
+pub fn init_routes(config: &mut web::ServiceConfig) {
+  config.service(
+    web::scope("airports")
+      .service(import_airports)
+      .service(get_airports)
+      .service(get_airport)
+      .service(create_airport)
+      .service(update_airport)
+      .service(delete_airports)
+      .service(delete_airport),
+  );
+}

api/src/auth/mod.rs

@@ -0,0 +1,58 @@
+use argon2::{
+  password_hash::{rand_core::OsRng, SaltString},
+  Argon2, PasswordHash, PasswordHasher, PasswordVerifier,
+};
+use rand::prelude::*;
+use rand_chacha::ChaCha20Rng;
+
+mod model;
+mod routes;
+mod session;
+
+pub use model::*;
+pub use session::*;
+pub use routes::init_routes;
+
+use crate::error::{ApiError, ApiResult};
+
+pub const SESSION_COOKIE_NAME: &str = "session";
+
+pub fn csprng_128bit(take: usize) -> String {
+  // Generate a CSPRNG 128-bit (16 byte) ID using alphanumeric characters (a-z, A-Z, 0-9)
+  let rng = ChaCha20Rng::from_entropy();
+  rng
+    .sample_iter(rand::distributions::Alphanumeric)
+    .take(take)
+    .map(char::from)
+    .collect()
+}
+
+pub fn hash(str: &str) -> ApiResult<String> {
+  let salt = SaltString::generate(&mut OsRng);
+  let bytes = str.as_bytes();
+  let hash = Argon2::default().hash_password(bytes, &salt)?.to_string();
+  Ok(hash)
+}
+
+pub fn verify_hash(str: &str, hash: &str) -> bool {
+  let bytes = str.as_bytes();
+  let parsed_hash = match PasswordHash::new(hash) {
+    Ok(h) => h,
+    Err(_) => return false,
+  };
+  match Argon2::default().verify_password(bytes, &parsed_hash) {
+    Ok(_) => true,
+    Err(_) => false,
+  }
+}
+
+pub fn verify_role(auth: &Auth, role: &str) -> ApiResult<()> {
+  if auth.user.role == role {
+    Ok(())
+  } else {
+    Err(ApiError {
+      status: 403,
+      message: "User does not have permission to perform this action.".to_string(),
+    })
+  }
+}

api/src/auth/model.rs

@@ -0,0 +1,67 @@
+use std::future::Future;
+use std::pin::Pin;
+
+use actix_web::{FromRequest, Error as ActixError, HttpRequest, dev::Payload, http};
+use serde::{Serialize, Deserialize};
+use crate::{
+  error::ApiError,
+  users::{User, UserResponse},
+};
+
+use super::{Session, SESSION_COOKIE_NAME};
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct Auth {
+  pub session_id: Option<String>,
+  pub user: UserResponse,
+}
+
+impl FromRequest for Auth {
+  type Error = ActixError;
+  type Future = Pin<Box<dyn Future<Output = Result<Self, Self::Error>>>>;
+
+  fn from_request(req: &HttpRequest, _: &mut Payload) -> Self::Future {
+    // Get session ID from request
+    let session_id = match req
+      .cookie(SESSION_COOKIE_NAME)
+      .map(|c| c.value().to_string())
+      .or_else(|| {
+        req
+          .headers()
+          .get(http::header::AUTHORIZATION)
+          .map(|h| h.to_str().unwrap().split_at(7).1.to_string())
+      }) {
+      Some(id) => id,
+      None => {
+        let fut = async {
+          Err(
+            ApiError {
+              status: 401,
+              message: "No session ID found in the request".to_string(),
+            }
+            .into(),
+          )
+        };
+        return Box::pin(fut);
+      }
+    };
+
+    // Get IP address from request
+    let ip_address = req.peer_addr().unwrap().ip().to_string();
+
+    // Verify the session
+    let fut = async move {
+      match Session::verify(&session_id, &ip_address).await {
+        Ok(session) => match User::get_by_email(&session.email) {
+          Ok(user) => Ok(Auth {
+            session_id: Some(session_id),
+            user: user.into(),
+          }),
+          Err(err) => Err(err.into()),
+        },
+        Err(err) => Err(err.into()),
+      }
+    };
+    Box::pin(fut)
+  }
+}

api/src/auth/routes.rs

@@ -0,0 +1,98 @@
+use actix_web::{
+  post, web, HttpResponse, ResponseError,
+  cookie::{Cookie, time::Duration},
+  HttpRequest,
+};
+use crate::{
+  auth::{verify_hash, Session, SESSION_COOKIE_NAME},
+  error::ApiError,
+  users::{LoginRequest, RegisterRequest, User, UserResponse},
+};
+
+use crate::auth::Auth;
+
+#[post("/register")]
+async fn register(user: web::Json<RegisterRequest>) -> HttpResponse {
+  let register_user = user.0;
+  let insert_user: User = match register_user.to_user() {
+    Ok(user) => user,
+    Err(err) => return ResponseError::error_response(&err),
+  };
+  match User::insert(insert_user) {
+    Ok(user) => {
+      let response: UserResponse = user.into();
+      HttpResponse::Created().json(response)
+    },
+    Err(err) => {
+      // Obfuscate the service error message to prevent leaking database details
+      if err.status == 409 {
+        return HttpResponse::Conflict().finish();
+      } else {
+        return ResponseError::error_response(&err);
+      }
+    }
+  }
+}
+
+#[post("/login")]
+async fn login(request: web::Json<LoginRequest>, req: HttpRequest) -> HttpResponse {
+  let email = request.email.clone();
+  let ip_address = req.peer_addr().unwrap().ip().to_string();
+
+  let query_user = match User::get_by_email(&email) {
+    Ok(query_user) => query_user,
+    Err(err) => {
+      log::error!("{}", err);
+      return ResponseError::error_response(&err);
+    }
+  };
+  if verify_hash(&query_user.hash, &request.password) {
+    // Create a session
+    let session = Session::new(&email, &ip_address);
+    let session_cookie = session.cookie();
+    // Save the session to the database
+    if let Err(err) = session.store().await {
+      log::error!("Failed to store session");
+      return ResponseError::error_response(&ApiError::new(500, err.to_string()));
+    }
+    return HttpResponse::Ok().cookie(session_cookie).finish();
+  } else {
+    log::error!("Invalid login attempt for {}", email);
+    return HttpResponse::Unauthorized().finish();
+  }
+}
+
+#[post("/logout")]
+async fn logout(req: HttpRequest, _auth: Auth) -> HttpResponse {
+  // Delete the session from the store
+  match req.cookie(SESSION_COOKIE_NAME) {
+    Some(cookie) => {
+      let session_id = cookie.value().to_string();
+      if let Err(err) = Session::delete(&session_id).await {
+        log::error!("Failed to delete session");
+        return ResponseError::error_response(&ApiError::new(500, err.to_string()));
+      }
+    }
+    None => {
+      return ResponseError::error_response(&ApiError::new(400, "Invalid session".to_string()));
+    }
+  }
+
+  let session_cookie = Cookie::build(SESSION_COOKIE_NAME, "")
+    .path("/")
+    .max_age(Duration::seconds(-1))
+    .secure(true)
+    .http_only(true)
+    .finish();
+
+  HttpResponse::Ok().cookie(session_cookie).finish()
+}
+
+pub fn init_routes(config: &mut web::ServiceConfig) {
+  config.service(
+    web::scope("auth")
+      .service(register)
+      .service(login)
+      .service(logout),
+  );
+}

api/src/auth/session.rs

@@ -0,0 +1,91 @@
+use actix_web::cookie::{time::Duration, Cookie};
+use chrono::{DateTime, Utc};
+use serde::{Deserialize, Serialize};
+use redis::{AsyncCommands, RedisResult};
+
+use crate::{
+  db::redis_async_connection,
+  error::{ApiError, ApiResult},
+};
+
+use super::{csprng_128bit, hash, verify_hash};
+
+pub const DEFAULT_SESSION_TTL: i64 = 86400; // (In seconds) 24 hours
+pub const SESSION_COOKIE_NAME: &str = "session";
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct Session {
+  pub session_id: String,
+  pub email: String,
+  pub ip_address: String,
+  pub expires_at: DateTime<Utc>,
+}
+
+impl Session {
+  pub fn new(email: &str, ip_address: &str) -> Self {
+    let now = chrono::Utc::now();
+    Self {
+      session_id: csprng_128bit(32),
+      email: email.to_string(),
+      ip_address: hash(&ip_address).unwrap(),
+      expires_at: now + chrono::Duration::seconds(DEFAULT_SESSION_TTL),
+    }
+  }
+
+  pub async fn store(&self) -> ApiResult<()> {
+    let mut conn = redis_async_connection().await?;
+    let key = self.session_id.clone();
+    let value = serde_json::to_string(self)?;
+    let result: RedisResult<()> = conn.set_ex(key, &value, DEFAULT_SESSION_TTL as u64).await;
+    match result {
+      Ok(_) => Ok(()),
+      Err(err) => Err(err.into()),
+    }
+  }
+
+  pub async fn get(session_id: &str) -> ApiResult<Option<Self>> {
+    let mut conn = redis_async_connection().await?;
+    let result: RedisResult<Option<String>> = conn.get(session_id).await;
+    match result {
+      Ok(Some(value)) => Ok(Some(serde_json::from_str(&value)?)),
+      Ok(None) => Ok(None),
+      Err(err) => Err(err.into()),
+    }
+  }
+
+  pub async fn delete(session_id: &str) -> ApiResult<()> {
+    let mut conn = redis_async_connection().await?;
+    let result: RedisResult<()> = conn.del(session_id).await;
+    match result {
+      Ok(_) => Ok(()),
+      Err(err) => Err(err.into()),
+    }
+  }
+
+  pub async fn verify(session_id: &str, ip_address: &str) -> ApiResult<Self> {
+    // Check if the session exists
+    let session = match Self::get(session_id).await? {
+      Some(session) => session,
+      None => return Err(ApiError::new(401, "Session does not exist".to_string())),
+    };
+
+    // Check if the IP Address matches the Session's IP Address
+    if verify_hash(ip_address, &session.ip_address) {
+      return Ok(session);
+    } else {
+      return Err(ApiError::new(
+        401,
+        "IP Address does not match".to_string(),
+      ));
+    }
+  }
+
+  pub fn cookie(&self) -> Cookie {
+    Cookie::build(SESSION_COOKIE_NAME, self.session_id.clone())
+      .path("/")
+      .max_age(Duration::seconds(DEFAULT_SESSION_TTL))
+      .secure(true)
+      .http_only(true)
+      .finish()
+  }
+}

api/src/db/mod.rs

@@ -0,0 +1,168 @@
+use crate::error::{ApiError, ApiResult};
+use diesel::{r2d2::ConnectionManager, PgConnection};
+use redis::{Client as RedisClient, aio::MultiplexedConnection as RedisConnection};
+use s3::{
+  Bucket, Region, creds::Credentials, BucketConfiguration, request::ResponseData,
+  bucket_ops::CreateBucketResponse,
+};
+use serde::{Deserialize, Serialize};
+use crate::diesel_migrations::MigrationHarness;
+use lazy_static::lazy_static;
+use log::{error, info, warn};
+use r2d2;
+use std::env;
+
+pub mod schema;
+
+type Pool = r2d2::Pool<ConnectionManager<PgConnection>>;
+pub type DbConnection = r2d2::PooledConnection<ConnectionManager<PgConnection>>;
+
+pub const MIGRATIONS: diesel_migrations::EmbeddedMigrations = embed_migrations!();
+
+lazy_static! {
+  static ref POOL: Pool = {
+    let username = env::var("DATABASE_USER").expect("Database username is not set");
+    let password = env::var("DATABASE_PASSWORD").expect("Database password is not set");
+    let host = env::var("DATABASE_HOST").unwrap_or("localhost".to_string());
+    let name = env::var("DATABASE_NAME").expect("Database name is not set");
+    let port = env::var("DATABASE_PORT").unwrap_or("5432".to_string());
+    let url = format!(
+      "postgres://{}:{}@{}:{}/{}",
+      username, password, host, port, name
+    );
+    let manager = ConnectionManager::<PgConnection>::new(url);
+    Pool::builder()
+      .test_on_check_out(true)
+      .build(manager)
+      .expect("Failed to create db pool")
+  };
+  static ref REDIS: RedisClient = {
+    let host = env::var("REDIS_HOST").unwrap_or("localhost".to_string());
+    let port = env::var("REDIS_PORT").unwrap_or("6379".to_string());
+    let url = format!("redis://{}:{}", host, port);
+    RedisClient::open(url).expect("Failed to create redis client")
+  };
+  static ref BUCKET: Bucket = {
+    let url = env::var("MINIO_HOST").unwrap_or("localhost".to_string());
+    let port = env::var("MINIO_PORT").unwrap_or("9000".to_string());
+    let user = env::var("MINIO_ROOT_USER").expect("MINIO_ROOT_USER is not set");
+    let password = env::var("MINIO_ROOT_PASSWORD").expect("MINIO_ROOT_PASSWORD is not set");
+    let base_url = format!("http://{}:{}", url, port);
+
+    let region = Region::Custom {
+      region: "".to_string(),
+      endpoint: base_url,
+    };
+
+    let credentials = Credentials {
+      access_key: Some(user),
+      secret_key: Some(password),
+      security_token: None,
+      session_token: None,
+      expiration: None,
+    };
+
+    *Bucket::new("aviation", region.clone(), credentials.clone())
+      .expect("Failed to create S3 Bucket")
+      .with_path_style()
+  };
+}
+
+pub async fn init() {
+  lazy_static::initialize(&POOL);
+  lazy_static::initialize(&REDIS);
+  lazy_static::initialize(&BUCKET);
+  match create_bucket().await {
+    Ok(_) => info!("Bucket initialized"),
+    Err(err) => match err.status {
+      409 => warn!("Bucket already exists"),
+      _ => error!("Failed to initialize bucket; {}", err),
+    },
+  };
+  let mut pool: DbConnection = connection().expect("Failed to get db connection");
+  match pool.run_pending_migrations(MIGRATIONS) {
+    Ok(_) => info!("Database initialized"),
+    Err(err) => error!("Failed to initialize database; {}", err),
+  };
+}
+
+pub fn connection() -> ApiResult<DbConnection> {
+  POOL
+    .get()
+    .map_err(|e| ApiError::new(500, format!("Failed getting db connection: {}", e)))
+}
+
+pub fn redis_connection() -> ApiResult<redis::Connection> {
+  let conn = REDIS.get_connection()?;
+  Ok(conn)
+}
+
+pub async fn redis_async_connection() -> ApiResult<RedisConnection> {
+  let conn = REDIS.get_multiplexed_async_connection().await?;
+  Ok(conn)
+}
+
+async fn create_bucket() -> ApiResult<CreateBucketResponse> {
+  let url = env::var("MINIO_URL").unwrap_or("localhost".to_string());
+  let port = env::var("MINIO_PORT").unwrap_or("9000".to_string());
+  let user = env::var("MINIO_ROOT_USER").expect("MINIO_ROOT_USER is not set");
+  let password = env::var("MINIO_ROOT_PASSWORD").expect("MINIO_ROOT_PASSWORD is not set");
+  let base_url = format!("http://{}:{}", url, port);
+
+  let region = Region::Custom {
+    region: "".to_string(),
+    endpoint: base_url,
+  };
+
+  let credentials = Credentials {
+    access_key: Some(user),
+    secret_key: Some(password),
+    security_token: None,
+    session_token: None,
+    expiration: None,
+  };
+  let bucket_name = "aviation";
+  let response = Bucket::create_with_path_style(
+    bucket_name,
+    region,
+    credentials,
+    BucketConfiguration::default(),
+  )
+  .await?;
+  Ok(response)
+}
+
+pub async fn upload_file(path: &str, content: &[u8]) -> ApiResult<ResponseData> {
+  let response = BUCKET.put_object(path, content).await?;
+  Ok(response)
+}
+
+pub async fn get_file(path: &str) -> ApiResult<Vec<u8>> {
+  let response = BUCKET.get_object(path).await?;
+  let bytes = response.bytes();
+  Ok(bytes.to_vec())
+}
+
+pub async fn delete_file(path: &str) -> ApiResult<ResponseData> {
+  let response = BUCKET.delete_object(path).await?;
+  Ok(response)
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct Response<T> {
+  pub data: T,
+  pub meta: Option<Metadata>,
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct Metadata {
+  pub page: i32,
+  pub limit: i32,
+  pub total: i64,
+}
+
+#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
+pub struct Coordinate {
+  pub lon: f64,
+  pub lat: f64,
+}

api/src/db/schema.rs

@@ -0,0 +1,41 @@
+diesel::table! {
+  use diesel::sql_types::*;
+  use postgis_diesel::sql_types::*;
+  airports (icao) {
+    icao -> Text,
+    category -> Text,
+    name -> Text,
+    elevation_ft -> Float,
+    iso_country -> Text,
+    iso_region -> Text,
+    municipality -> Text,
+    has_metar -> Bool,
+    point -> Geometry,
+    data -> Jsonb
+  }
+}
+
+diesel::table! {
+  metars (id) {
+    id -> Integer,
+    icao -> Text,
+    observation_time -> Timestamp,
+    raw_text -> Text,
+    data -> Jsonb,
+  }
+}
+
+diesel::table! {
+  users (email) {
+    email -> Text,
+    hash -> Text,
+    role -> Text,
+    first_name -> Text,
+    last_name -> Text,
+    updated_at -> Timestamp,
+    created_at -> Timestamp,
+    profile_picture -> Nullable<Text>,
+    favorites -> Array<Text>,
+    verified -> Bool,
+  }
+}

api/src/error.rs

@@ -0,0 +1,148 @@
+use actix_web::http::StatusCode;
+use actix_web::{HttpResponse, ResponseError};
+use diesel::result::Error as DieselError;
+use log::warn;
+use regex::Regex;
+use serde::{Deserialize, Serialize};
+use serde_json::json;
+use std::fmt;
+
+pub type ApiResult<T> = Result<T, ApiError>;
+
+#[derive(Debug, Deserialize, Serialize)]
+pub struct ApiError {
+  pub status: u16,
+  pub message: String,
+}
+
+impl ApiError {
+  pub fn new(status: u16, message: String) -> Self {
+    Self { status, message }
+  }
+
+  pub fn to_http_response(&self) -> HttpResponse {
+    let status = match StatusCode::from_u16(self.status) {
+      Ok(s) => s,
+      Err(err) => {
+        warn!("{}", err);
+        StatusCode::INTERNAL_SERVER_ERROR
+      }
+    };
+    HttpResponse::build(status).body(self.message.to_string())
+  }
+}
+
+impl fmt::Display for ApiError {
+  fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+    f.write_str(self.message.as_str())
+  }
+}
+
+impl From<std::io::Error> for ApiError {
+  fn from(error: std::io::Error) -> Self {
+    Self::new(500, format!("Unknown IO error: {}", error))
+  }
+}
+
+impl From<std::env::VarError> for ApiError {
+  fn from(error: std::env::VarError) -> Self {
+    Self::new(
+      500,
+      format!("Unknown environment variable error: {}", error),
+    )
+  }
+}
+
+impl From<DieselError> for ApiError {
+  fn from(error: DieselError) -> Self {
+    match error {
+      DieselError::DatabaseError(kind, err) => match kind {
+        diesel::result::DatabaseErrorKind::UniqueViolation => {
+          Self::new(409, err.message().to_string())
+        }
+        _ => Self::new(500, err.message().to_string()),
+      },
+      DieselError::NotFound => Self::new(404, "The record was not found".to_string()),
+      DieselError::SerializationError(err) => Self::new(422, err.to_string()),
+      err => Self::new(500, format!("Unknown Diesel error: {}", err)),
+    }
+  }
+}
+
+impl From<reqwest::Error> for ApiError {
+  fn from(error: reqwest::Error) -> Self {
+    Self::new(500, format!("Unknown reqwest error: {}", error))
+  }
+}
+
+impl From<serde_json::Error> for ApiError {
+  fn from(error: serde_json::Error) -> Self {
+    Self::new(500, format!("Unknown serde_json error: {}", error))
+  }
+}
+
+impl From<argon2::password_hash::Error> for ApiError {
+  fn from(error: argon2::password_hash::Error) -> Self {
+    Self::new(500, format!("Unknown argon2 error: {}", error))
+  }
+}
+
+impl From<redis::RedisError> for ApiError {
+  fn from(error: redis::RedisError) -> Self {
+    Self::new(500, format!("Unknown redis error: {}", error))
+  }
+}
+
+impl From<s3::error::S3Error> for ApiError {
+  fn from(error: s3::error::S3Error) -> Self {
+    match error {
+      s3::error::S3Error::Credentials(err) => {
+        Self::new(500, format!("Unknown s3 credentials error: {}", err))
+      }
+      s3::error::S3Error::FromUtf8(err) => {
+        Self::new(500, format!("Unknown s3 from utf8 error: {}", err))
+      }
+      s3::error::S3Error::FmtError(err) => {
+        Self::new(500, format!("Unknown s3 fmt error: {}", err))
+      }
+      s3::error::S3Error::HeaderToStr(err) => {
+        Self::new(500, format!("Unknown s3 header to str error: {}", err))
+      }
+      s3::error::S3Error::HmacInvalidLength(err) => Self::new(
+        500,
+        format!("Unknown s3 hmac invalid length error: {}", err),
+      ),
+      s3::error::S3Error::Http(error) => {
+        Self::new(error.status_code().as_u16(), error.to_string())
+      }
+      _ => {
+        let re = Regex::new(r"HTTP (\d{3})").unwrap();
+        // Apply the regex to the input string
+        if let Some(captures) = re.captures(&error.to_string()) {
+          if let Some(http_code_str) = captures.get(1) {
+            if let Ok(http_code) = http_code_str.as_str().parse::<u16>() {
+              return Self::new(http_code, error.to_string());
+            }
+          }
+        }
+        Self::new(500, format!("Unknown s3 error: {}", error))
+      }
+    }
+  }
+}
+
+impl ResponseError for ApiError {
+  fn error_response(&self) -> HttpResponse {
+    let status = match StatusCode::from_u16(self.status) {
+      Ok(status) => status,
+      Err(_) => StatusCode::INTERNAL_SERVER_ERROR,
+    };
+
+    let message = match status.as_u16() < 500 {
+      true => self.message.clone(),
+      false => "Internal server error".to_string(),
+    };
+
+    HttpResponse::build(status).json(json!({ "status": status.as_u16(), "message": message }))
+  }
+}

api/src/main.rs

@@ -0,0 +1,57 @@
+extern crate diesel;
+#[macro_use]
+extern crate diesel_migrations;
+
+use std::env;
+
+use actix_cors::Cors;
+use actix_web::{App, HttpServer, middleware::Logger};
+use dotenv::dotenv;
+
+mod airports;
+mod auth;
+mod db;
+mod error;
+mod metars;
+mod scheduler;
+mod users;
+
+#[actix_web::main]
+async fn main() -> std::io::Result<()> {
+  dotenv().ok();
+  env_logger::init_from_env(env_logger::Env::default().filter_or("RUST_LOG", "warn,api=info"));
+  db::init().await;
+  // scheduler::update_airports();
+
+  let host = env::var("API_HOST").unwrap_or("localhost".to_string());
+  let port = env::var("API_PORT").unwrap_or("5000".to_string());
+
+  let server = match HttpServer::new(move || {
+    let cors = Cors::default()
+      .allow_any_origin()
+      .allow_any_method()
+      .allow_any_header()
+      .supports_credentials()
+      .max_age(3600);
+    App::new()
+      .wrap(cors)
+      .wrap(Logger::default())
+      .configure(airports::init_routes)
+      .configure(metars::init_routes)
+      .configure(auth::init_routes)
+      .configure(users::init_routes)
+  })
+  .bind(format!("{}:{}", host, port))
+  {
+    Ok(b) => {
+      log::info!("Binding server to {}:{}", host, port);
+      b
+    }
+    Err(err) => {
+      log::error!("Could not bind server: {}", err);
+      return Err(err);
+    }
+  };
+
+  server.run().await
+}

api/src/metars/mod.rs

@@ -0,0 +1,5 @@
+mod model;
+mod routes;
+
+pub use model::*;
+pub use routes::init_routes;

api/src/metars/model.rs

@@ -0,0 +1,888 @@
+use crate::airports::QueryAirport;
+use crate::error::ApiError;
+use crate::{error::ApiResult, db};
+use crate::db::schema::metars::{self};
+use chrono::Datelike;
+use diesel::{prelude::*, sql_query};
+use log::{warn, trace};
+use std::collections::HashSet;
+use serde::{Deserialize, Serialize};
+
+#[derive(Serialize, Deserialize, Debug)]
+pub struct QualityControlFlags {
+  #[serde(skip_serializing_if = "Option::is_none")]
+  pub auto: Option<bool>,
+  #[serde(skip_serializing_if = "Option::is_none")]
+  pub auto_station_without_precipication: Option<bool>,
+  #[serde(skip_serializing_if = "Option::is_none")]
+  pub auto_station_with_precipication: Option<bool>,
+  #[serde(skip_serializing_if = "Option::is_none")]
+  pub maintenance_indicator_on: Option<bool>,
+  #[serde(skip_serializing_if = "Option::is_none")]
+  pub corrected: Option<bool>,
+  #[serde(skip_serializing_if = "Option::is_none")]
+  pub no_significant_change: Option<bool>,
+  #[serde(skip_serializing_if = "Option::is_none")]
+  pub temporary_change: Option<bool>,
+}
+
+impl Default for QualityControlFlags {
+  fn default() -> Self {
+    QualityControlFlags {
+      auto: None,
+      auto_station_without_precipication: None,
+      auto_station_with_precipication: None,
+      maintenance_indicator_on: None,
+      corrected: None,
+      no_significant_change: None,
+      temporary_change: None,
+    }
+  }
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+pub struct SkyCondition {
+  pub sky_cover: String,
+  #[serde(skip_serializing_if = "Option::is_none")]
+  pub cloud_base_ft_agl: Option<i32>,
+  #[serde(skip_serializing_if = "Option::is_none")]
+  pub significant_convective_clouds: Option<String>,
+}
+
+impl Default for SkyCondition {
+  fn default() -> Self {
+    SkyCondition {
+      sky_cover: "".to_string(),
+      cloud_base_ft_agl: None,
+      significant_convective_clouds: None,
+    }
+  }
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+pub struct RunwayVisualRange {
+  pub runway: String,
+  #[serde(skip_serializing_if = "Option::is_none")]
+  pub visibility_ft: Option<String>,
+  #[serde(skip_serializing_if = "Option::is_none")]
+  pub variable_visibility_high_ft: Option<String>,
+  #[serde(skip_serializing_if = "Option::is_none")]
+  pub variable_visibility_low_ft: Option<String>,
+}
+
+impl Default for RunwayVisualRange {
+  fn default() -> Self {
+    RunwayVisualRange {
+      runway: "".to_string(),
+      visibility_ft: None,
+      variable_visibility_high_ft: None,
+      variable_visibility_low_ft: None,
+    }
+  }
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+pub enum FlightCategory {
+  VFR,
+  MVFR,
+  LIFR,
+  IFR,
+  UNKN,
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+pub struct Metar {
+  pub raw_text: String,
+  pub station_id: String,
+  pub observation_time: chrono::NaiveDateTime,
+  #[serde(skip_serializing_if = "Option::is_none")]
+  pub temp_c: Option<f64>,
+  #[serde(skip_serializing_if = "Option::is_none")]
+  pub dewpoint_c: Option<f64>,
+  #[serde(skip_serializing_if = "Option::is_none")]
+  pub wind_dir_degrees: Option<String>,
+  #[serde(skip_serializing_if = "Option::is_none")]
+  pub wind_speed_kt: Option<f64>,
+  #[serde(skip_serializing_if = "Option::is_none")]
+  pub wind_gust_kt: Option<f64>,
+  #[serde(skip_serializing_if = "Option::is_none")]
+  pub variable_wind_dir_degrees: Option<String>,
+  #[serde(skip_serializing_if = "Option::is_none")]
+  pub visibility_statute_mi: Option<String>,
+  pub runway_visual_range: Vec<RunwayVisualRange>,
+  #[serde(skip_serializing_if = "Option::is_none")]
+  pub altim_in_hg: Option<f64>,
+  #[serde(skip_serializing_if = "Option::is_none")]
+  pub sea_level_pressure_mb: Option<f64>,
+  pub quality_control_flags: QualityControlFlags,
+  pub weather_phenomena: Vec<String>,
+  pub sky_condition: Vec<SkyCondition>,
+  pub flight_category: FlightCategory,
+  #[serde(skip_serializing_if = "Option::is_none")]
+  pub three_hr_pressure_tendency_mb: Option<f64>,
+  #[serde(skip_serializing_if = "Option::is_none")]
+  pub max_t_c: Option<f64>,
+  #[serde(skip_serializing_if = "Option::is_none")]
+  pub min_t_c: Option<f64>,
+  #[serde(skip_serializing_if = "Option::is_none")]
+  pub precip_in: Option<f64>,
+}
+
+impl Default for Metar {
+  fn default() -> Self {
+    Metar {
+      raw_text: "".to_string(),
+      station_id: "".to_string(),
+      observation_time: chrono::NaiveDateTime::parse_from_str(
+        "1970-01-01T00:00:00",
+        "%Y-%m-%dT%H:%M:%S",
+      )
+      .unwrap(),
+      temp_c: None,
+      dewpoint_c: None,
+      wind_dir_degrees: None,
+      wind_speed_kt: None,
+      wind_gust_kt: None,
+      variable_wind_dir_degrees: None,
+      visibility_statute_mi: None,
+      runway_visual_range: vec![],
+      altim_in_hg: None,
+      sea_level_pressure_mb: None,
+      quality_control_flags: QualityControlFlags::default(),
+      weather_phenomena: vec![],
+      sky_condition: vec![],
+      flight_category: FlightCategory::UNKN,
+      three_hr_pressure_tendency_mb: None,
+      max_t_c: None,
+      min_t_c: None,
+      precip_in: None,
+    }
+  }
+}
+
+impl Metar {
+  fn parse(metar_strings: Vec<&str>) -> ApiResult<Vec<Self>> {
+    let mut metars: Vec<Self> = vec![];
+    for metar_string in metar_strings {
+      trace!("Parsing METAR data: {}", metar_string);
+      let mut metar: Metar = Metar::default();
+      metar.raw_text = metar_string.to_owned();
+      let mut metar_parts: Vec<&str> = metar_string.split_whitespace().collect();
+      if metar_parts.len() < 4 {
+        warn!(
+          "Unable to parse METAR data in an unexpected format: {}",
+          metar_string
+        );
+        continue;
+      }
+
+      // Station Identifier
+      metar.station_id = metar_parts[0].to_string();
+      metar_parts.remove(0);
+
+      // Date/Time
+      let observation_time = metar_parts[0];
+      metar_parts.remove(0);
+      if observation_time.len() != 7 {
+        warn!(
+          "Unable to parse observation time in {}: {}",
+          observation_time, metar_string
+        );
+        continue;
+      }
+      let observation_time_day = &observation_time[0..2];
+      let observation_time_hour = &observation_time[2..4];
+      let observation_time_minute = &observation_time[4..6];
+      let current_time = chrono::Utc::now().naive_utc();
+      // Check if the observation time is from the previous month
+      let observation_time_month =
+        if current_time.day() > observation_time_day.parse::<u32>().unwrap() {
+          current_time.month() - 1
+        } else {
+          current_time.month()
+        };
+      // Check if the observation time is from the previous year
+      let observation_time_year = if current_time.month() > observation_time_month {
+        current_time.year() - 1
+      } else {
+        current_time.year()
+      };
+      // Handle Daylight Savings Time
+      let observation_time_hour =
+        if observation_time_month == 3 && observation_time_day.parse::<u32>().unwrap() < 14 {
+          observation_time_hour.parse::<u32>().unwrap() - 1
+        } else {
+          observation_time_hour.parse::<u32>().unwrap()
+        };
+      let observation_time = format!(
+        "{}-{}-{}T{}:{}:00Z",
+        observation_time_year,
+        observation_time_month,
+        observation_time_day,
+        observation_time_hour,
+        observation_time_minute
+      );
+      metar.observation_time =
+        chrono::NaiveDateTime::parse_from_str(&observation_time, "%Y-%m-%dT%H:%M:%SZ").unwrap();
+
+      loop {
+        if metar_parts.is_empty() {
+          break;
+        }
+        // Report Modifiers
+        if !metar_parts.is_empty() && metar_parts[0] == "AUTO" {
+          metar.quality_control_flags.auto = Some(true);
+          metar_parts.remove(0);
+        }
+        if !metar_parts.is_empty() && metar_parts[0] == "COR" {
+          metar.quality_control_flags.corrected = Some(true);
+          metar_parts.remove(0);
+        }
+        if !metar_parts.is_empty() && metar_parts[0] == "NOSIG" {
+          metar.quality_control_flags.no_significant_change = Some(true);
+          metar_parts.remove(0);
+        }
+
+        // Wind Direction and Speed
+        let wind_re = regex::Regex::new(r"^(?:[0-9]{3}|VRB)[0-9]{2}(?:KT|MPS)$").unwrap();
+        let wind_gust_re =
+          regex::Regex::new(r"^(?:[0-9]{3}|VRB)[0-9]{2}G[0-9]{2}(?:KT|MPS)$").unwrap();
+        // Handle input error where there is a space between the numbers and units
+        let mut value: Option<String> = None;
+        if metar_parts.len() >= 2
+          && metar_parts[0].len() == 5
+          && (metar_parts[1] == "KT" || metar_parts[1] == "MPS")
+        {
+          value = Some(format!("{}{}", metar_parts[0], metar_parts[1]));
+          metar_parts.remove(0);
+          metar_parts.remove(0);
+        } else if metar_parts.len() >= 2
+          && metar_parts[0].len() == 7
+          && metar_parts[0].contains("G")
+          && (metar_parts[1] == "KT" || metar_parts[1] == "MPS")
+        {
+          value = Some(format!("{}{}", metar_parts[0], metar_parts[1]));
+          metar_parts.remove(0);
+          metar_parts.remove(0);
+        } else if !metar_parts.is_empty() && wind_re.is_match(metar_parts[0]) {
+          value = Some(metar_parts[0].to_string());
+          metar_parts.remove(0);
+        } else if !metar_parts.is_empty() && wind_gust_re.is_match(metar_parts[0]) {
+          value = Some(metar_parts[0].to_string());
+          metar_parts.remove(0);
+        }
+
+        match value {
+          Some(wind) => {
+            if wind_re.is_match(&wind) {
+              let wind_dir_degrees = &wind[0..3];
+              metar.wind_dir_degrees = Some(wind_dir_degrees.to_string());
+              let mut wind_speed_kt = wind[3..5].to_string();
+              // Convert m/s to kt
+              if wind.len() == 8 {
+                wind_speed_kt = (wind_speed_kt.parse::<f64>().unwrap() * 1.94384).to_string();
+              }
+              metar.wind_speed_kt = Some(wind_speed_kt.parse::<f64>().unwrap());
+            } else if wind_gust_re.is_match(&wind) {
+              let wind_dir_degrees = &wind[0..3];
+              metar.wind_dir_degrees = Some(wind_dir_degrees.to_string());
+              let mut wind_speed_kt = wind[3..5].to_string();
+              let mut wind_gust_kt = wind[6..8].to_string();
+              // Convert m/s to kt
+              if wind.len() == 9 {
+                wind_speed_kt = (wind_speed_kt.parse::<f64>().unwrap() * 1.94384).to_string();
+                wind_gust_kt = (wind_gust_kt.parse::<f64>().unwrap() * 1.94384).to_string();
+              }
+              metar.wind_speed_kt = Some(wind_speed_kt.parse::<f64>().unwrap());
+              metar.wind_gust_kt = Some(wind_gust_kt.parse::<f64>().unwrap());
+            }
+          }
+          None => {}
+        }
+
+        // Variable Wind Direction
+        let variable_wind_re = regex::Regex::new(r"^[0-9]{3}V[0-9]{3}$").unwrap();
+        if !metar_parts.is_empty() && variable_wind_re.is_match(metar_parts[0]) {
+          metar.variable_wind_dir_degrees = Some(metar_parts[0].to_string());
+          metar_parts.remove(0);
+        }
+
+        // Visibility
+        let visibility_re = regex::Regex::new(r"^M?(?:[0-9]+|[0-9]+/[0-9]+)SM$").unwrap();
+        let visibility_re_m = regex::Regex::new(r"^[0-9]{4}(:?N|NE|NW|S|SE|SW)?$").unwrap();
+        if !metar_parts.is_empty() && visibility_re.is_match(metar_parts[0]) {
+          let visibility_str = &metar_parts[0][0..metar_parts[0].len() - 2];
+          metar_parts.remove(0);
+          let visibility: String = if visibility_str.contains("/") {
+            let visibility_parts: Vec<&str> = visibility_str.split("/").collect();
+            let visibility_left = visibility_parts[0];
+            let visibility_right = visibility_parts[1].parse::<f64>().unwrap();
+            if visibility_left.starts_with("M") {
+              format!(
+                "M{}",
+                visibility_left[1..visibility_left.len()]
+                  .parse::<f64>()
+                  .unwrap()
+                  / visibility_right
+              )
+            } else if visibility_left.starts_with("P") {
+              format!(
+                "P{}",
+                visibility_left[1..visibility_left.len()]
+                  .parse::<f64>()
+                  .unwrap()
+                  / visibility_right
+              )
+            } else {
+              format!(
+                "{}",
+                visibility_left.parse::<f64>().unwrap() / visibility_right
+              )
+            }
+          } else {
+            visibility_str.to_string()
+          };
+          metar.visibility_statute_mi = Some(visibility);
+        } else if !metar_parts.is_empty()
+          && metar_parts[0].parse::<f64>().is_ok()
+          && metar_parts.len() > 1
+          && visibility_re.is_match(metar_parts[1])
+        {
+          let visibility_whole = metar_parts[0].parse::<f64>().unwrap();
+          metar_parts.remove(0);
+          let visibility_parts: Vec<&str> = metar_parts[0].split("/").collect();
+          metar_parts.remove(0);
+          let visibility_left = visibility_parts[0];
+          let visibility_right = visibility_parts[1][0..visibility_parts[1].len() - 2]
+            .parse::<f64>()
+            .unwrap();
+          let visibility = if visibility_left.starts_with("M") {
+            format!(
+              "M{}",
+              visibility_whole
+                + (visibility_left[1..visibility_left.len()]
+                  .parse::<f64>()
+                  .unwrap()
+                  / visibility_right)
+            )
+          } else if visibility_left.starts_with("P") {
+            format!(
+              "P{}",
+              visibility_whole
+                + (visibility_left[1..visibility_left.len()]
+                  .parse::<f64>()
+                  .unwrap()
+                  / visibility_right)
+            )
+          } else {
+            format!(
+              "{}",
+              visibility_whole + (visibility_left.parse::<f64>().unwrap() / visibility_right)
+            )
+          };
+          metar.visibility_statute_mi = Some(visibility);
+        } else if !metar_parts.is_empty() && visibility_re_m.is_match(metar_parts[0]) {
+          // Convert meters to statute miles
+          let visibility = metar_parts[0];
+          metar_parts.remove(0);
+          if &visibility[0..4] == "9999" {
+            metar.visibility_statute_mi = Some("P10".to_string());
+          } else {
+            let visibility = visibility[0..4].parse::<f64>().unwrap() * 0.000621371;
+            metar.visibility_statute_mi = Some(format!("{:.2}", visibility));
+          }
+        }
+
+        // Runway Visual Range
+        let rvr_re = regex::Regex::new(r"^R[0-9]{1,3}(?:L|R|C)?/[PM]?[0-9]{4}FT$").unwrap();
+        let variable_rvr_re =
+          regex::Regex::new(r"^R[0-9]{1,3}(?:L|R|C)?/[PM]?[0-9]{4}V[PM]?[0-9]{4}FT$").unwrap();
+        while !metar_parts.is_empty()
+          && (rvr_re.is_match(metar_parts[0]) || variable_rvr_re.is_match(metar_parts[0]))
+        {
+          let rvr_string = metar_parts[0];
+          metar_parts.remove(0);
+          let mut rvr = RunwayVisualRange::default();
+          let rvr_parts: Vec<&str> = rvr_string.split("/").collect();
+          rvr.runway = rvr_parts[0].to_string();
+          if rvr_re.is_match(rvr_string) {
+            rvr.visibility_ft = Some(rvr_parts[1].to_string());
+          } else {
+            let rvr_variable_parts: Vec<&str> = rvr_parts[1].split("V").collect();
+            if rvr_variable_parts.len() != 2 {
+              warn!(
+                "Unable to parse runway visual range in {}: {}",
+                rvr_string, metar_string
+              );
+            } else {
+              rvr.variable_visibility_low_ft = Some(rvr_variable_parts[0].to_string());
+              rvr.variable_visibility_high_ft = Some(rvr_variable_parts[1].to_string());
+            }
+          }
+        }
+
+        // Weather Phenomena
+        let wx_intensity = "(?:[+-]|VC)?";
+        let wx_descriptor = "(?:MI|PR|BC|DR|BL|SH|TS|FZ)?";
+        let wx_precipitation =
+          "(?:DZ|RA|SN|SG|IC|PL|GR|GS|UP|BR|FG|FU|VA|DU|SA|HZ|PY|PO|SQ|FC|SS|DS)?";
+        let wx_re = regex::Regex::new(&format!(
+          r"^{}{}{}$",
+          wx_intensity, wx_descriptor, wx_precipitation
+        ))
+        .unwrap();
+        while !metar_parts.is_empty() && wx_re.is_match(metar_parts[0]) {
+          metar.weather_phenomena.push(metar_parts[0].to_string());
+          metar_parts.remove(0);
+        }
+
+        // Sky Condition
+        if !metar_parts.is_empty() && metar_parts[0] == "CAVOK" {
+          metar.sky_condition.push(SkyCondition {
+            sky_cover: "CLR".to_string(),
+            cloud_base_ft_agl: None,
+            significant_convective_clouds: None,
+          });
+          metar_parts.remove(0);
+        }
+        let sky_condition_re =
+          regex::Regex::new(r"^(?:CLR|SKC|NSC|NCD|(?:FEW|SCT|BKN|OVC|VV)([0-9/]{3})?(?:CB|TCU)?)$")
+            .unwrap();
+        while !metar_parts.is_empty() && sky_condition_re.is_match(metar_parts[0]) {
+          let sky_condition_string = metar_parts[0];
+          metar_parts.remove(0);
+          let mut sky_condition = SkyCondition::default();
+          let mut vv_offset = 0;
+          if &sky_condition_string[0..2] == "VV" {
+            sky_condition.sky_cover = "VV".to_string();
+            vv_offset = 1;
+          } else {
+            sky_condition.sky_cover = sky_condition_string[0..3].to_string();
+          }
+          if sky_condition_string.len() > 3 - vv_offset {
+            // Parse out the next three digits
+            let cloud_base_ft_agl = &sky_condition_string[3 - vv_offset..6 - vv_offset];
+            if cloud_base_ft_agl == "///" {
+              sky_condition.cloud_base_ft_agl = None;
+            } else {
+              sky_condition.cloud_base_ft_agl = match cloud_base_ft_agl.parse::<i32>() {
+                Ok(c) => Some(c * 100),
+                Err(err) => {
+                  warn!(
+                    "Unable to parse cloud base in {}: {}",
+                    sky_condition_string, err
+                  );
+                  None
+                }
+              };
+            }
+            if sky_condition_string.len() > 6 - vv_offset {
+              // Parse out the next two digits
+              let scc = &sky_condition_string[6 - vv_offset..8 - vv_offset];
+              sky_condition.significant_convective_clouds = Some(scc.to_string());
+            }
+          }
+          metar.sky_condition.push(sky_condition);
+        }
+
+        // Temperature and Dewpoint
+        let temp_re = regex::Regex::new(r"^(?:M?[0-9]{2})?/(?:M?[0-9]{2})?$").unwrap();
+        if !metar_parts.is_empty() && temp_re.is_match(metar_parts[0]) {
+          let temp_string = metar_parts[0];
+          metar_parts.remove(0);
+          let temp_parts: Vec<&str> = temp_string.split("/").collect();
+          let mut temp_c = "";
+          let mut dewpoint_c = "";
+          if temp_parts.len() != 2 {
+            if temp_string.ends_with("/") {
+              temp_c = temp_parts[0];
+            } else {
+              dewpoint_c = temp_parts[0];
+            }
+          } else {
+            temp_c = temp_parts[0];
+            dewpoint_c = temp_parts[1];
+          }
+          if temp_c.starts_with("M") {
+            metar.temp_c = Some(temp_c[1..temp_c.len()].parse::<f64>().unwrap() * -1.0);
+          } else if !temp_c.is_empty() {
+            metar.temp_c = match temp_c.parse::<f64>() {
+              Ok(t) => Some(t),
+              Err(err) => {
+                warn!("Unable to parse temperature in {}: {}", temp_c, err);
+                None
+              }
+            };
+          }
+          if dewpoint_c.starts_with("M") {
+            metar.dewpoint_c = Some(dewpoint_c[1..dewpoint_c.len()].parse::<f64>().unwrap() * -1.0);
+          } else if !dewpoint_c.is_empty() {
+            metar.dewpoint_c = match dewpoint_c.parse::<f64>() {
+              Ok(d) => Some(d),
+              Err(err) => {
+                warn!("Unable to parse dewpoint in {}: {}", dewpoint_c, err);
+                None
+              }
+            };
+          }
+        }
+
+        // Altimeter
+        let altim_re = regex::Regex::new(r"^A[0-9]{4}$").unwrap();
+        if !metar_parts.is_empty() && altim_re.is_match(metar_parts[0]) {
+          let altim = metar_parts[0];
+          metar_parts.remove(0);
+          metar.altim_in_hg = Some(altim[1..altim.len()].parse::<f64>().unwrap() / 100.0);
+        }
+
+        // Pressure
+        let pressure_re = regex::Regex::new(r"^Q[0-9]{4}$").unwrap();
+        if !metar_parts.is_empty() && pressure_re.is_match(metar_parts[0]) {
+          let pressure = metar_parts[0];
+          metar_parts.remove(0);
+          metar.sea_level_pressure_mb = Some(pressure[1..pressure.len()].parse::<f64>().unwrap());
+        }
+
+        // Temporary Change
+        if !metar_parts.is_empty() && metar_parts[0] == "TEMPO" {
+          metar.quality_control_flags.temporary_change = Some(true);
+          metar_parts.remove(0);
+        }
+
+        // Remarks
+        if !metar_parts.is_empty() && metar_parts[0] == "RMK" {
+          metar_parts.remove(0);
+          loop {
+            if metar_parts.is_empty() {
+              break;
+            }
+            let slp_re = regex::Regex::new(r"^SLP([0-9]{3})$").unwrap();
+            let hourly_temp_re = regex::Regex::new(r"^T[01][0-9]{3}[01][0-9]{3}$").unwrap();
+            let remark = metar_parts[0];
+            metar_parts.remove(0);
+            if remark == "AO1" {
+              metar
+                .quality_control_flags
+                .auto_station_without_precipication = Some(true);
+            } else if remark == "AO2" {
+              metar.quality_control_flags.auto_station_with_precipication = Some(true);
+            } else if remark == "$" {
+              metar.quality_control_flags.maintenance_indicator_on = Some(true);
+            } else if slp_re.is_match(remark) {
+              let slp = slp_re.captures(remark).unwrap();
+              let sea_level_pressure = slp[1].parse::<f64>().unwrap();
+              if sea_level_pressure > 500.0 {
+                metar.sea_level_pressure_mb = Some((sea_level_pressure / 10.0) + 900.0);
+              } else {
+                metar.sea_level_pressure_mb = Some((sea_level_pressure / 10.0) + 1000.0);
+              }
+            } else if hourly_temp_re.is_match(remark) {
+              let temp_negation = &remark[1..2];
+              let temp = &remark[2..5];
+              if let Ok(t) = temp.parse::<f64>() {
+                if temp_negation == "0" {
+                  metar.temp_c = Some(t / 10.0);
+                } else {
+                  metar.temp_c = Some(t / 10.0 * -1.0);
+                }
+              }
+              let dewpoint_negation = &remark[6..7];
+              let dewpoint = &remark[6..9];
+              if let Ok(d) = dewpoint.parse::<f64>() {
+                if dewpoint_negation == "0" {
+                  metar.dewpoint_c = Some(d / 10.0);
+                } else {
+                  metar.dewpoint_c = Some(d / 10.0 * -1.0);
+                }
+              }
+            }
+          }
+        }
+
+        // Skip unexpected fields
+        if !metar_parts.is_empty() {
+          warn!(
+            "Skipping unexpected field: '{}' ({})",
+            metar_parts[0], metar_string
+          );
+          metar_parts.remove(0);
+        }
+      }
+
+      // Flight Category
+      if metar.visibility_statute_mi.is_none() && metar.sky_condition.is_empty() {
+        metar.flight_category = FlightCategory::UNKN;
+      } else {
+        let visibility = match &metar.visibility_statute_mi {
+          Some(v) => {
+            if v.starts_with("M") || v.starts_with("P") {
+              v[1..v.len()].parse::<f64>().unwrap()
+            } else {
+              v.parse::<f64>().unwrap()
+            }
+          }
+          None => 5.0, // Assume VFR if no visibility is present
+        };
+        // Ceiling is the lowest cloud base that is BKN or OVC
+        let ceiling = match metar.sky_condition.first() {
+          Some(s) => {
+            if s.sky_cover == "VV" {
+              0.0
+            } else if s.sky_cover == "BKN" || s.sky_cover == "OVC" {
+              match s.cloud_base_ft_agl {
+                Some(c) => c as f64,
+                None => 0.0,
+              }
+            } else {
+              3000.0 // Assume VFR if no BKN or OVC sky condition is present
+            }
+          }
+          None => 3000.0, // Assume VFR if no sky condition is present
+        };
+        if visibility >= 5.0 && ceiling >= 3000.0 {
+          metar.flight_category = FlightCategory::VFR;
+        } else if visibility >= 3.0 && ceiling >= 1000.0 {
+          metar.flight_category = FlightCategory::MVFR;
+        } else if visibility >= 1.0 && ceiling >= 500.0 {
+          metar.flight_category = FlightCategory::IFR;
+        } else {
+          metar.flight_category = FlightCategory::LIFR;
+        }
+      }
+
+      metars.push(metar);
+    }
+    return Ok(metars);
+  }
+
+  fn get_missing_metar_icaos(db_metars: &Vec<Self>, station_icaos: &Vec<&str>) -> Vec<String> {
+    let mut missing_metar_icaos: Vec<String> = vec![];
+    let current_time = chrono::Local::now().naive_local().and_utc().timestamp();
+    let db_metars_set: HashSet<&str> = db_metars
+      .iter()
+      .map(|icao| icao.station_id.as_str())
+      .collect();
+    let station_icaos_set: HashSet<&str> = station_icaos.to_owned().into_iter().collect();
+    for difference in db_metars_set.symmetric_difference(&station_icaos_set) {
+      missing_metar_icaos.push(difference.to_string());
+    }
+    for metar in db_metars {
+      if current_time > (metar.observation_time.and_utc().timestamp() + 3600) {
+        trace!("{} METAR data is outdated", metar.station_id);
+        missing_metar_icaos.push(metar.station_id.to_string());
+      }
+    }
+    return missing_metar_icaos;
+  }
+
+  async fn get_remote_metars(icaos: Vec<String>) -> ApiResult<Vec<Metar>> {
+    let gov_api_url = std::env::var("GOV_API_URL").expect("GOV_API_URL must be set");
+    // Query the remote API for the missing METAR data 10 at a time
+    let icao_chunks = icaos
+      .chunks(10)
+      .map(|chunk| chunk.join(","))
+      .collect::<Vec<String>>();
+    let mut metars: Vec<Metar> = vec![];
+    for icao_chunk in icao_chunks {
+      let url = format!("{}/metar.php?ids={}", gov_api_url, icao_chunk);
+      let mut m = match reqwest::get(url).await {
+        Ok(r) => {
+          // Check if the status code is 200
+          if r.status() != 200 {
+            return Err(ApiError::new(
+              500,
+              format!("Unable to get METAR request: {}", r.status()),
+            ));
+          }
+          match r.text().await {
+            Ok(r) => {
+              let metar_chunk = r
+                .trim()
+                .split("\n")
+                .filter(|m| !m.trim().is_empty())
+                .collect();
+              match Metar::parse(metar_chunk) {
+                Ok(m) => m,
+                Err(err) => return Err(err),
+              }
+            }
+            Err(err) => {
+              return Err(ApiError::new(
+                500,
+                format!("Unable to parse METAR request: {}", err),
+              ))
+            }
+          }
+        }
+        Err(err) => {
+          return Err(ApiError::new(
+            500,
+            format!("Unable to get METAR request: {}", err),
+          ))
+        }
+      };
+      metars.append(&mut m);
+    }
+    return Ok(metars);
+  }
+
+  fn from_query(query_metars: Vec<QueryMetar>) -> Vec<Self> {
+    let mut metars: Vec<Metar> = vec![];
+    for metar in query_metars {
+      let mut metar: Metar = serde_json::from_value(metar.data).unwrap();
+      metar.raw_text = metar.raw_text.to_string();
+      metar.station_id = metar.station_id.to_string();
+      metars.push(metar);
+    }
+    return metars;
+  }
+
+  fn to_insert(metars: &Vec<Self>) -> Vec<InsertMetar> {
+    let mut insert_metars: Vec<InsertMetar> = vec![];
+    for metar in metars {
+      insert_metars.push(InsertMetar {
+        icao: metar.station_id.to_string(),
+        observation_time: metar.observation_time,
+        raw_text: metar.raw_text.to_string(),
+        data: serde_json::to_value(metar).unwrap(),
+      });
+    }
+    return insert_metars;
+  }
+
+  pub async fn get_all(icao_string: String) -> ApiResult<Vec<Self>> {
+    if icao_string.is_empty() {
+      return Ok(vec![]);
+    }
+
+    let icaos: Vec<&str> = icao_string.split(",").collect();
+
+    let mut db_metars = match QueryMetar::get_all(&icaos) {
+      Ok(m) => Self::from_query(m),
+      Err(err) => return Err(err),
+    };
+
+    let missing_icaos = Self::get_missing_metar_icaos(&db_metars, &icaos);
+    if missing_icaos.is_empty() {
+      return Ok(db_metars);
+    }
+    trace!("Retrieving missing METAR data for {:?}", missing_icaos);
+    let missing_icaos_string: Vec<String> = missing_icaos
+      .iter()
+      .map(|icao| format!("{}", icao.to_string()))
+      .collect();
+    let mut airports: Vec<QueryAirport> = vec![];
+    missing_icaos_string
+      .clone()
+      .iter()
+      .for_each(|icao| match QueryAirport::get(icao) {
+        Ok(a) => airports.push(a),
+        Err(_) => {}
+      });
+    let missing_result = Self::get_remote_metars(missing_icaos_string).await;
+    let mut missing_metars = match missing_result {
+      Ok(m) => m,
+      Err(err) => {
+        warn!("Unable to get remote METAR data; {}", err);
+        vec![]
+      }
+    };
+    if missing_metars.len() > 0 {
+      let insert_metars = Self::to_insert(&missing_metars);
+      match InsertMetar::insert(&insert_metars) {
+        Ok(rows) => trace!("Inserted {} metar rows", rows),
+        Err(err) => warn!("Unable to insert metar data; {}", err),
+      };
+      // Update airports with the appropriate has_metar flag
+      airports.iter().for_each(|airport| {
+        if missing_metars
+          .iter()
+          .any(|metar| metar.station_id == airport.icao)
+        {
+          let updated = QueryAirport {
+            icao: airport.icao.to_string(),
+            category: airport.category.to_string(),
+            name: airport.name.to_string(),
+            elevation_ft: airport.elevation_ft,
+            iso_country: airport.iso_country.to_string(),
+            iso_region: airport.iso_region.to_string(),
+            municipality: airport.municipality.to_string(),
+            has_metar: true,
+            point: airport.point,
+            data: airport.data.to_owned(),
+          };
+          match QueryAirport::update(updated) {
+            Ok(_) => {}
+            Err(err) => warn!("Unable to update airport with has_metar flag; {}", err),
+          }
+        }
+      });
+    }
+    let mut metars: Vec<Metar> = vec![];
+    metars.append(&mut missing_metars);
+    metars.append(&mut db_metars);
+    Ok(metars)
+  }
+}
+
+#[derive(Serialize, Deserialize, AsChangeset, Insertable)]
+#[diesel(table_name = metars)]
+struct InsertMetar {
+  icao: String,
+  observation_time: chrono::NaiveDateTime,
+  raw_text: String,
+  data: serde_json::Value,
+}
+
+impl InsertMetar {
+  fn insert(metars: &Vec<Self>) -> ApiResult<usize> {
+    let mut conn = db::connection()?;
+    match diesel::insert_into(metars::table)
+      .values(metars)
+      .execute(&mut conn)
+    {
+      Ok(rows) => Ok(rows),
+      Err(err) => Err(ApiError {
+        status: 500,
+        message: format!("{}", err),
+      }),
+    }
+  }
+}
+
+#[derive(Serialize, Deserialize, Queryable, QueryableByName)]
+#[diesel(table_name = metars)]
+struct QueryMetar {
+  id: i32,
+  icao: String,
+  observation_time: chrono::NaiveDateTime,
+  raw_text: String,
+  data: serde_json::Value,
+}
+
+impl QueryMetar {
+  fn get_all(icaos: &Vec<&str>) -> ApiResult<Vec<QueryMetar>> {
+    // Sanitize search to only allow [a-zA-Z0-9]
+    let icaos = icaos
+      .iter()
+      .map(|icao| {
+        icao
+          .chars()
+          .filter(|c| c.is_alphanumeric())
+          .collect::<String>()
+      })
+      .collect::<Vec<String>>();
+    let station_query: Vec<String> = icaos
+      .iter()
+      .map(|icao| format!("'{}'", icao.to_string()))
+      .collect();
+    let mut conn = db::connection()?;
+    let db_metars: Vec<Self> = match sql_query(
+      format!("SELECT DISTINCT ON (icao) * FROM metars WHERE icao IN ({}) ORDER BY icao, observation_time DESC", station_query.join(","))
+    ).load(&mut conn) {
+      Ok(m) => m,
+      Err(err) => return Err(ApiError { status: 500, message: format!("{}", err) })
+    };
+    return Ok(db_metars);
+  }
+}

api/src/metars/routes.rs

@@ -0,0 +1,45 @@
+use crate::{error::ApiError, db::Metadata};
+use crate::metars::Metar;
+use actix_web::{get, web, HttpResponse, HttpRequest};
+use log::error;
+use serde::{Deserialize, Serialize};
+
+#[derive(Serialize, Deserialize)]
+pub struct MetarsResponse {
+  pub data: Vec<Metar>,
+  pub meta: Metadata,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct GetAllParameters {
+  icaos: Option<String>,
+}
+
+#[get("metars")]
+async fn get_all(req: HttpRequest) -> HttpResponse {
+  let params = web::Query::<GetAllParameters>::from_query(req.query_string()).unwrap();
+  let icao_option = params.icaos.clone();
+  let icao_string = match icao_option {
+    Some(i) => i,
+    None => return HttpResponse::UnprocessableEntity().body("Missing icaos parameter"),
+  };
+
+  let metars =
+    match web::block(|| Ok::<_, ApiError>(async { Metar::get_all(icao_string).await }))
+      .await
+      .unwrap()
+      .unwrap()
+      .await
+    {
+      Ok(a) => a,
+      Err(err) => {
+        error!("{}", err);
+        return err.to_http_response();
+      }
+    };
+  HttpResponse::Ok().json(metars)
+}
+
+pub fn init_routes(config: &mut web::ServiceConfig) {
+  config.service(get_all);
+}

api/src/scheduler.rs

@@ -0,0 +1,74 @@
+use tokio::time::{sleep, Duration};
+
+use crate::airports::{QueryAirport, QueryFilters};
+use crate::metars::Metar;
+
+pub fn update_airports() {
+  tokio::spawn(async {
+    let mut airports: Vec<QueryAirport> = vec![];
+    let limit = 100;
+    loop {
+      log::debug!("METAR update start");
+      let total = match QueryAirport::get_count(&QueryFilters::default()) {
+        Ok(t) => t,
+        Err(err) => {
+          log::warn!("{}", err);
+          break;
+        }
+      };
+      if total != airports.len() as i64 {
+        log::debug!("{} cached airports, expected {}", airports.len(), total);
+        airports = vec![];
+        let pages = ((total as f32) / (if limit <= 0 { 1 } else { limit } as f32)).ceil() as i32;
+        for page in 1..(pages + 1) {
+          match QueryAirport::get_all(&QueryFilters::default(), limit, page) {
+            Ok(mut a) => airports.append(&mut a),
+            Err(err) => {
+              log::warn!("{}", err);
+              break;
+            }
+          }
+        }
+      }
+      log::debug!("Updating {} airport METARS", airports.len());
+
+      let airport_icaos: Vec<String> = airports.iter().map(|a| a.icao.to_string()).collect();
+      let mut peekable = airport_icaos.into_iter().peekable();
+      let mut observation_time = chrono::Utc::now().timestamp();
+
+      if peekable.peek().is_none() {
+        log::debug!("No airports to update, sleeping for 1 hour");
+        sleep(Duration::from_secs(3600)).await;
+        continue;
+      }
+
+      while peekable.peek().is_some() {
+        let chunk: Vec<String> = peekable.by_ref().take(limit as usize).collect();
+        let icao_string = chunk.join(",");
+        log::trace!("Updating METARS for: {}", icao_string);
+        match Metar::get_all(icao_string).await {
+          Ok(metars) => {
+            // Find the oldest observation time
+            for metar in metars {
+              if metar.observation_time.and_utc().timestamp() < observation_time {
+                observation_time = metar.observation_time.and_utc().timestamp();
+              }
+            }
+          }
+          Err(err) => {
+            log::warn!("{}", err);
+          }
+        }
+        // Sleep for 100ms between chunks to avoid rate limiting
+        sleep(Duration::from_millis(100)).await;
+      }
+      log::debug!("METAR update complete");
+      // Sleep until the earliest observation time is 1 hour old
+      // Bounded by 1 and 3600 seconds
+      let now = chrono::Utc::now().timestamp();
+      let sleep_time = std::cmp::min(std::cmp::max(1, now - (observation_time + 3600)), 3600);
+      log::debug!("Next update in {} seconds", sleep_time);
+      sleep(Duration::from_secs(sleep_time as u64)).await;
+    }
+  });
+}

api/src/users/mod.rs

@@ -0,0 +1,5 @@
+mod model;
+mod routes;
+
+pub use model::*;
+pub use routes::init_routes;

api/src/users/model.rs

@@ -0,0 +1,108 @@
+use serde::{Deserialize, Serialize};
+use diesel::prelude::*;
+
+use crate::{
+  auth::hash,
+  db::{connection, schema::users},
+  error::ApiResult,
+};
+
+/**
+ * RegisterRequest
+ */
+#[derive(Debug, Serialize, Deserialize)]
+pub struct RegisterRequest {
+  pub email: String,
+  pub password: String,
+  pub first_name: String,
+  pub last_name: String,
+}
+
+impl RegisterRequest {
+  pub fn to_user(self) -> ApiResult<User> {
+    let hash = hash(&self.password)?;
+    Ok(User {
+      email: self.email.to_lowercase(),
+      hash,
+      role: "user".to_string(),
+      first_name: self.first_name,
+      last_name: self.last_name,
+      updated_at: chrono::Utc::now().naive_utc(),
+      created_at: chrono::Utc::now().naive_utc(),
+      profile_picture: None,
+      favorites: vec![],
+      verified: false,
+    })
+  }
+}
+
+/**
+ * LoginRequest
+ */
+#[derive(Debug, Serialize, Deserialize)]
+pub struct LoginRequest {
+  pub email: String,
+  pub password: String,
+}
+
+/**
+ * UserResponse
+ */
+#[derive(Debug, Serialize, Deserialize)]
+pub struct UserResponse {
+  pub email: String,
+  pub role: String,
+  pub first_name: String,
+  pub last_name: String,
+  pub profile_picture: Option<String>,
+}
+
+impl From<User> for UserResponse {
+  fn from(user: User) -> Self {
+    UserResponse {
+      email: user.email,
+      role: user.role,
+      first_name: user.first_name,
+      last_name: user.last_name,
+      profile_picture: user.profile_picture,
+    }
+  }
+}
+
+/**
+ * User
+ */
+#[derive(Debug, Insertable, AsChangeset, Queryable, QueryableByName, Serialize, Deserialize)]
+#[diesel(table_name = users)]
+pub struct User {
+  pub email: String,
+  pub hash: String,
+  pub role: String,
+  pub first_name: String,
+  pub last_name: String,
+  pub updated_at: chrono::NaiveDateTime,
+  pub created_at: chrono::NaiveDateTime,
+  pub profile_picture: Option<String>,
+  pub favorites: Vec<String>,
+  pub verified: bool,
+}
+
+impl User {
+  pub fn get_by_email(email: &str) -> ApiResult<User> {
+    let mut conn = connection()?;
+    // Check if the user exists by email, case insensitive
+
+    let user = users::table
+      .filter(users::email.eq(email.to_lowercase()))
+      .first(&mut conn)?;
+    Ok(user)
+  }
+
+  pub fn insert(user: Self) -> ApiResult<User> {
+    let mut conn = connection()?;
+    let user = diesel::insert_into(users::table)
+      .values(user)
+      .get_result(&mut conn)?;
+    Ok(user)
+  }
+}

api/src/users/routes.rs

@@ -0,0 +1,165 @@
+// use actix_multipart::Multipart;
+// use actix_web::{get, post, delete, web, HttpResponse, ResponseError};
+// use futures_util::StreamExt;
+
+// use crate::{
+//   auth::Auth,
+//   db::{delete_file, get_file, upload_file},
+//   error::ServiceError,
+//   users::User,
+// };
+
+// #[get("/favorites")]
+// async fn get_favorites(auth: Auth) -> HttpResponse {
+//   match User::get_by_email(&auth.user.email) {
+//     Ok(user) => return HttpResponse::Ok().json(user.favorites),
+//     Err(err) => return ResponseError::error_response(&err),
+//   }
+// }
+
+// #[post("/favorites/{icao}")]
+// async fn add_favorite(icao: web::Path<String>, auth: Auth) -> HttpResponse {
+//   match User::get_by_email(&auth.user.email) {
+//     Ok(user) => {
+//       if user.favorites.contains(&icao) {
+//         // Check if the airport ICAO is already in the user's favorites
+//         return HttpResponse::Conflict().finish();
+//       } else {
+//         // Add the airport ICAO to the user's favorites
+//         let mut favorites = user.favorites;
+//         favorites.push(icao.into_inner());
+//         match User::update_favorites(&user.email, favorites) {
+//           Ok(_) => return HttpResponse::Ok().finish(),
+//           Err(err) => return ResponseError::error_response(&err),
+//         }
+//       }
+//     }
+//     Err(err) => return ResponseError::error_response(&err),
+//   }
+// }
+
+// #[delete("/favorites/{icao}")]
+// async fn delete_favorite(icao: web::Path<String>, auth: Auth) -> HttpResponse {
+//   let icao: String = icao.into_inner();
+//   match User::get_by_email(&auth.user.email) {
+//     Ok(user) => {
+//       if user.favorites.contains(&icao) {
+//         // Check if the airport ICAO is already in the user's favorites
+//         let mut favorites = user.favorites;
+//         favorites.retain(|x| x != &icao);
+//         match User::update_favorites(&user.email, favorites) {
+//           Ok(_) => return HttpResponse::Ok().finish(),
+//           Err(err) => return ResponseError::error_response(&err),
+//         }
+//       } else {
+//         // Remove the airport ICAO from the user's favorites
+//         return HttpResponse::Conflict().finish();
+//       }
+//     }
+//     Err(err) => return ResponseError::error_response(&err),
+//   }
+// }
+
+// #[post("/picture")]
+// async fn set_picture(mut payload: Multipart, auth: Auth) -> HttpResponse {
+//   while let Some(item) = payload.next().await {
+//     let mut bytes = web::BytesMut::new();
+//     let mut field = match item {
+//       Ok(field) => field,
+//       Err(err) => return ResponseError::error_response(&err),
+//     };
+//     let content_type = field.content_disposition();
+//     let filename = match content_type.unwrap().get_filename() {
+//       Some(name) => match name.split(".").last() {
+//         Some(ext) => match ext {
+//           "apng" | "avif" | "gif" | "jpg" | "jpeg" | "jfif" | "pjpeg" | "pjp" | "png" | "svg"
+//           | "webp" => name,
+//           _ => {
+//             return ResponseError::error_response(&ServiceError::new(
+//               400,
+//               "File extension is not supported".to_string(),
+//             ))
+//           }
+//         },
+//         None => {
+//           return ResponseError::error_response(&ServiceError::new(
+//             400,
+//             "Unknown file extension".to_string(),
+//           ))
+//         }
+//       },
+//       None => {
+//         return ResponseError::error_response(&ServiceError::new(
+//           400,
+//           "File name is not provided".to_string(),
+//         ))
+//       }
+//     };
+//     let path = format!("users/{}/{}", auth.user.email, filename);
+
+//     while let Some(chunk) = field.next().await {
+//       let data = match chunk {
+//         Ok(data) => data,
+//         Err(err) => return ResponseError::error_response(&err),
+//       };
+//       bytes.extend_from_slice(&data);
+//     }
+//     match upload_file(&path, &bytes).await {
+//       Ok(_) => {
+//         match User::update_profile_picture(&auth.user.email, Some(&path)) {
+//           Ok(_) => {}
+//           Err(err) => return ResponseError::error_response(&err),
+//         };
+//       }
+//       Err(err) => return ResponseError::error_response(&err),
+//     };
+//   }
+//   HttpResponse::Ok().finish()
+// }
+
+// #[get("/picture")]
+// async fn get_picture(auth: Auth) -> HttpResponse {
+//   let user = match User::get_by_email(&auth.user.email) {
+//     Ok(user) => user,
+//     Err(err) => return ResponseError::error_response(&err),
+//   };
+//   if let Some(path) = user.profile_picture {
+//     match get_file(&path).await {
+//       Ok(bytes) => HttpResponse::Ok().body(bytes),
+//       Err(err) => ResponseError::error_response(&err),
+//     }
+//   } else {
+//     HttpResponse::NotFound().finish()
+//   }
+// }
+
+// #[delete("/picture")]
+// async fn delete_picture(auth: Auth) -> HttpResponse {
+//   let user = match User::get_by_email(&auth.user.email) {
+//     Ok(user) => user,
+//     Err(err) => return ResponseError::error_response(&err),
+//   };
+//   if let Some(path) = user.profile_picture {
+//     match delete_file(&path).await {
+//       Ok(_) => match User::update_profile_picture(&auth.user.email, None) {
+//         Ok(_) => HttpResponse::Ok().finish(),
+//         Err(err) => ResponseError::error_response(&err),
+//       },
+//       Err(err) => ResponseError::error_response(&err),
+//     }
+//   } else {
+//     HttpResponse::NotFound().finish()
+//   }
+// }
+
+pub fn init_routes(config: &mut actix_web::web::ServiceConfig) {
+  // config.service(
+  //   web::scope("users")
+  //     .service(get_favorites)
+  //     .service(add_favorite)
+  //     .service(delete_favorite)
+  //     .service(set_picture)
+  //     .service(get_picture)
+  //     .service(delete_picture),
+  // );
+}