diff --git a/Cargo.toml b/Cargo.toml index c4fd484..12afb8e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -94,6 +94,7 @@ url = "2.5" urlencoding = "2.1" base32 = "0.5.1" futures = { version = "0.3.31", default-features = false, features = ["alloc"] } +h3o = "0.6" [profile.release] opt-level = 3 diff --git a/lexicon/events.smokesignal.lfg.json b/lexicon/events.smokesignal.lfg.json new file mode 100644 index 0000000..6de52af --- /dev/null +++ b/lexicon/events.smokesignal.lfg.json @@ -0,0 +1,52 @@ +{ + "lexicon": 1, + "id": "events.smokesignal.lfg", + "defs": { + "main": { + "type": "record", + "description": "A Looking For Group record that broadcasts interest in finding activity partners within a geographic area.", + "key": "tid", + "record": { + "type": "object", + "required": ["location", "tags", "startsAt", "endsAt", "createdAt", "active"], + "properties": { + "location": { + "type": "ref", + "ref": "community.lexicon.location#geo", + "description": "The geographic location for activity partner matching." + }, + "tags": { + "type": "array", + "description": "Interest tags for matching with events and other users.", + "items": { + "type": "string", + "maxLength": 64, + "maxGraphemes": 64 + }, + "minLength": 1, + "maxLength": 10 + }, + "startsAt": { + "type": "string", + "format": "datetime", + "description": "When the LFG becomes active." + }, + "endsAt": { + "type": "string", + "format": "datetime", + "description": "When the LFG expires and is no longer visible." + }, + "createdAt": { + "type": "string", + "format": "datetime", + "description": "Record creation timestamp." + }, + "active": { + "type": "boolean", + "description": "Whether the LFG is currently active and visible to others." + } + } + } + } + } +} diff --git a/src/atproto/lexicon/lfg.rs b/src/atproto/lexicon/lfg.rs new file mode 100644 index 0000000..84279a2 --- /dev/null +++ b/src/atproto/lexicon/lfg.rs @@ -0,0 +1,304 @@ +//! Looking For Group (LFG) lexicon implementation. +//! +//! This module defines the LFG record structure for AT Protocol storage. +//! LFG records allow users to broadcast their interest in finding activity +//! partners within a geographic area for a limited time period. + +use atproto_record::lexicon::community::lexicon::location::LocationOrRef; +use atproto_record::typed::{LexiconType, TypedLexicon}; +use chrono::{DateTime, Utc}; +use h3o::{CellIndex, LatLng, Resolution}; +use serde::{Deserialize, Serialize}; + +/// Minimum H3 precision allowed for LFG locations (inclusive) +pub const MIN_H3_PRECISION: u8 = 4; + +/// Maximum H3 precision allowed for LFG locations (inclusive) +pub const MAX_H3_PRECISION: u8 = 7; + +pub const NSID: &str = "events.smokesignal.lfg"; + +/// Maximum number of tags allowed per LFG record +pub const MAX_TAGS: usize = 10; + +/// Maximum length of each tag in characters +pub const MAX_TAG_LENGTH: usize = 64; + +/// A Looking For Group record that broadcasts interest in finding activity +/// partners within a geographic area. +#[derive(Clone, Serialize, Deserialize, PartialEq)] +#[serde(rename_all = "camelCase")] +pub struct Lfg { + /// The geographic location for activity partner matching + pub location: LocationOrRef, + + /// Interest tags for matching with events and other users (1-10 items) + pub tags: Vec, + + /// When the LFG becomes active + pub starts_at: DateTime, + + /// When the LFG expires and is no longer visible + pub ends_at: DateTime, + + /// Record creation timestamp + pub created_at: DateTime, + + /// Whether the LFG is currently active and visible to others + pub active: bool, +} + +pub type TypedLfg = TypedLexicon; + +impl LexiconType for Lfg { + fn lexicon_type() -> &'static str { + NSID + } +} + +impl Lfg { + /// Validates the LFG record + pub fn validate(&self) -> Result<(), String> { + // Validate tags + if self.tags.is_empty() { + return Err("At least one tag is required".to_string()); + } + if self.tags.len() > MAX_TAGS { + return Err(format!("Maximum {} tags allowed", MAX_TAGS)); + } + for tag in &self.tags { + if tag.trim().is_empty() { + return Err("Tags cannot be empty".to_string()); + } + if tag.len() > MAX_TAG_LENGTH { + return Err(format!( + "Tag '{}' exceeds maximum length of {} characters", + tag, MAX_TAG_LENGTH + )); + } + } + + // Validate time range + if self.ends_at <= self.starts_at { + return Err("End time must be after start time".to_string()); + } + + // Validate location is an H3 cell with appropriate precision + match &self.location { + LocationOrRef::InlineHthree(h3) => { + // Parse the H3 cell index + let cell: CellIndex = h3 + .inner + .value + .parse() + .map_err(|_| format!("Invalid H3 cell: {}", h3.inner.value))?; + + // Check precision is within allowed range (4-7) + let resolution: Resolution = cell.resolution(); + let precision = u8::from(resolution); + + if precision < MIN_H3_PRECISION || precision > MAX_H3_PRECISION { + return Err(format!( + "H3 precision must be between {} and {} (got {})", + MIN_H3_PRECISION, MAX_H3_PRECISION, precision + )); + } + } + _ => { + return Err(format!( + "LFG location must be an inline H3 cell with precision {}-{}", + MIN_H3_PRECISION, MAX_H3_PRECISION + )); + } + } + + Ok(()) + } + + /// Extract latitude and longitude from the H3 cell center + pub fn get_coordinates(&self) -> Option<(f64, f64)> { + match &self.location { + LocationOrRef::InlineHthree(h3) => { + let cell: CellIndex = h3.inner.value.parse().ok()?; + let lat_lng: LatLng = cell.into(); + Some((lat_lng.lat(), lat_lng.lng())) + } + _ => None, + } + } + + /// Get the H3 cell index from the location + pub fn get_h3_cell(&self) -> Option { + match &self.location { + LocationOrRef::InlineHthree(h3) => h3.inner.value.parse().ok(), + _ => None, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use atproto_record::lexicon::community::lexicon::location::{Hthree, TypedHthree}; + use chrono::Duration; + + /// Create a test H3 cell at resolution 6 (New York area) + /// Resolution 6 cells are ~36km edge length + fn create_test_h3() -> LocationOrRef { + // H3 cell at resolution 6 covering part of New York + // This is a valid H3 index at resolution 6 + LocationOrRef::InlineHthree(TypedHthree::new(Hthree { + value: "862a1072fffffff".to_string(), + name: Some("New York Area".to_string()), + })) + } + + /// Create a test H3 cell at resolution 3 (too low precision) + fn create_test_h3_low_precision() -> LocationOrRef { + LocationOrRef::InlineHthree(TypedHthree::new(Hthree { + value: "832a10fffffffff".to_string(), + name: None, + })) + } + + /// Create a test H3 cell at resolution 8 (too high precision) + fn create_test_h3_high_precision() -> LocationOrRef { + LocationOrRef::InlineHthree(TypedHthree::new(Hthree { + value: "882a1072a9fffff".to_string(), + name: None, + })) + } + + fn create_test_lfg() -> Lfg { + let now = Utc::now(); + Lfg { + location: create_test_h3(), + tags: vec!["hiking".to_string(), "outdoors".to_string()], + starts_at: now, + ends_at: now + Duration::hours(48), + created_at: now, + active: true, + } + } + + #[test] + fn test_lfg_serialization() { + let lfg = create_test_lfg(); + let serialized = serde_json::to_string(&lfg).unwrap(); + + assert!(serialized.contains("\"startsAt\"")); + assert!(serialized.contains("\"endsAt\"")); + assert!(serialized.contains("\"createdAt\"")); + assert!(serialized.contains("\"active\":true")); + assert!(serialized.contains("\"tags\":[\"hiking\",\"outdoors\"]")); + } + + #[test] + fn test_lfg_deserialization() { + let lfg = create_test_lfg(); + let serialized = serde_json::to_string(&lfg).unwrap(); + let deserialized: Lfg = serde_json::from_str(&serialized).unwrap(); + + assert_eq!(deserialized.tags, lfg.tags); + assert_eq!(deserialized.active, lfg.active); + } + + #[test] + fn test_lfg_validation_valid() { + let lfg = create_test_lfg(); + assert!(lfg.validate().is_ok()); + } + + #[test] + fn test_lfg_validation_no_tags() { + let mut lfg = create_test_lfg(); + lfg.tags = vec![]; + assert!(lfg.validate().is_err()); + assert_eq!( + lfg.validate().unwrap_err(), + "At least one tag is required" + ); + } + + #[test] + fn test_lfg_validation_too_many_tags() { + let mut lfg = create_test_lfg(); + lfg.tags = (0..11).map(|i| format!("tag{}", i)).collect(); + assert!(lfg.validate().is_err()); + assert!(lfg.validate().unwrap_err().contains("Maximum 10 tags")); + } + + #[test] + fn test_lfg_validation_empty_tag() { + let mut lfg = create_test_lfg(); + lfg.tags = vec!["hiking".to_string(), " ".to_string()]; + assert!(lfg.validate().is_err()); + assert!(lfg.validate().unwrap_err().contains("empty")); + } + + #[test] + fn test_lfg_validation_invalid_time_range() { + let mut lfg = create_test_lfg(); + lfg.ends_at = lfg.starts_at - Duration::hours(1); + assert!(lfg.validate().is_err()); + assert!(lfg.validate().unwrap_err().contains("End time")); + } + + #[test] + fn test_lfg_get_coordinates() { + let lfg = create_test_lfg(); + let coords = lfg.get_coordinates(); + assert!(coords.is_some()); + let (lat, lon) = coords.unwrap(); + // H3 cell center coordinates (approximate - within the general NY area) + assert!(lat > 40.0 && lat < 42.0, "Latitude should be in NY area"); + assert!(lon > -75.0 && lon < -73.0, "Longitude should be in NY area"); + } + + #[test] + fn test_lfg_get_h3_cell() { + let lfg = create_test_lfg(); + let cell = lfg.get_h3_cell(); + assert!(cell.is_some()); + let cell = cell.unwrap(); + assert_eq!(u8::from(cell.resolution()), 6); + } + + #[test] + fn test_lfg_validation_h3_precision_too_low() { + let mut lfg = create_test_lfg(); + lfg.location = create_test_h3_low_precision(); + let result = lfg.validate(); + assert!(result.is_err()); + assert!(result.unwrap_err().contains("precision must be between")); + } + + #[test] + fn test_lfg_validation_h3_precision_too_high() { + let mut lfg = create_test_lfg(); + lfg.location = create_test_h3_high_precision(); + let result = lfg.validate(); + assert!(result.is_err()); + assert!(result.unwrap_err().contains("precision must be between")); + } + + #[test] + fn test_lfg_validation_non_h3_location() { + use atproto_record::lexicon::community::lexicon::location::{Geo, TypedGeo}; + + let mut lfg = create_test_lfg(); + lfg.location = LocationOrRef::InlineGeo(TypedGeo::new(Geo { + latitude: "40.7128".to_string(), + longitude: "-74.0060".to_string(), + name: Some("New York".to_string()), + })); + let result = lfg.validate(); + assert!(result.is_err()); + assert!(result.unwrap_err().contains("must be an inline H3 cell")); + } + + #[test] + fn test_lexicon_type() { + assert_eq!(Lfg::lexicon_type(), "events.smokesignal.lfg"); + } +} diff --git a/src/atproto/lexicon/mod.rs b/src/atproto/lexicon/mod.rs index e113059..87a4214 100644 --- a/src/atproto/lexicon/mod.rs +++ b/src/atproto/lexicon/mod.rs @@ -1,3 +1,4 @@ pub mod acceptance; pub mod bluesky_profile; +pub mod lfg; pub mod profile; diff --git a/src/bin/smokesignal.rs b/src/bin/smokesignal.rs index 2b6cdfe..19142c9 100644 --- a/src/bin/smokesignal.rs +++ b/src/bin/smokesignal.rs @@ -336,6 +336,7 @@ async fn main() -> Result<()> { match SearchIndexer::new( opensearch_endpoint, + pool.clone(), identity_resolver.clone(), document_storage.clone(), ) diff --git a/src/http/auth_utils.rs b/src/http/auth_utils.rs index 84bd901..6ddaff9 100644 --- a/src/http/auth_utils.rs +++ b/src/http/auth_utils.rs @@ -1,7 +1,9 @@ use anyhow::Result; use atproto_client::client::get_dpop_json_with_headers; +use deadpool_redis::redis::AsyncCommands; use http::HeaderMap; use reqwest::Client; +use sha2::{Digest, Sha256}; use crate::atproto::auth::create_dpop_auth_from_aip_session; use crate::config::OAuthBackendConfig; @@ -10,6 +12,9 @@ use crate::http::errors::LoginError; use crate::http::errors::web_error::WebError; use crate::http::middleware_auth::Auth; +/// TTL for AIP session ready cache entries (5 minutes). +const AIP_SESSION_READY_CACHE_TTL_SECS: i64 = 300; + /// Result of checking if an AIP session is ready for AT Protocol operations. pub(crate) enum AipSessionStatus { /// Session is valid and ready for operations. @@ -82,9 +87,82 @@ pub(crate) async fn check_aip_session_ready( } } +/// Generate a cache key for AIP session ready status. +/// +/// Uses a SHA-256 hash of the access token to avoid storing raw tokens in Redis. +fn aip_session_cache_key(access_token: &str) -> String { + let mut hasher = Sha256::new(); + hasher.update(access_token.as_bytes()); + let hash = hasher.finalize(); + format!("aip_session_ready:{:x}", hash) +} + +/// Check Redis cache for AIP session ready status. +/// +/// Returns `Some(true)` if cached as ready, `Some(false)` if cached as stale, +/// or `None` on cache miss or error. +async fn get_cached_aip_session_status( + web_context: &WebContext, + access_token: &str, +) -> Option { + let cache_key = aip_session_cache_key(access_token); + + let mut conn = match web_context.cache_pool.get().await { + Ok(conn) => conn, + Err(e) => { + tracing::debug!(?e, "Failed to get Redis connection for AIP session cache"); + return None; + } + }; + + match conn.get::<_, Option>(&cache_key).await { + Ok(Some(value)) => { + tracing::debug!(cache_key = %cache_key, value = %value, "AIP session cache hit"); + Some(value == "ready") + } + Ok(None) => { + tracing::debug!(cache_key = %cache_key, "AIP session cache miss"); + None + } + Err(e) => { + tracing::debug!(?e, "Redis error reading AIP session cache"); + None + } + } +} + +/// Cache AIP session ready status in Redis with TTL. +async fn cache_aip_session_status(web_context: &WebContext, access_token: &str, is_ready: bool) { + let cache_key = aip_session_cache_key(access_token); + let value = if is_ready { "ready" } else { "stale" }; + + let mut conn = match web_context.cache_pool.get().await { + Ok(conn) => conn, + Err(e) => { + tracing::debug!(?e, "Failed to get Redis connection for AIP session cache write"); + return; + } + }; + + if let Err(e) = conn + .set_ex::<_, _, ()>(&cache_key, value, AIP_SESSION_READY_CACHE_TTL_SECS as u64) + .await + { + tracing::debug!(?e, "Failed to cache AIP session status"); + } else { + tracing::debug!( + cache_key = %cache_key, + value = %value, + ttl_secs = AIP_SESSION_READY_CACHE_TTL_SECS, + "Cached AIP session status" + ); + } +} + /// Check if the current AIP session is ready for AT Protocol operations. /// /// This calls the AIP ready endpoint to validate the access token. +/// Results are cached in Redis for 5 minutes to avoid repeated validation. /// For PDS sessions, this is a no-op (returns NotAip). pub(crate) async fn require_valid_aip_session( web_context: &WebContext, @@ -101,6 +179,17 @@ pub(crate) async fn require_valid_aip_session( return Ok(AipSessionStatus::Stale); } + // Check Redis cache first + if let Some(cached_ready) = + get_cached_aip_session_status(web_context, access_token).await + { + return if cached_ready { + Ok(AipSessionStatus::Ready) + } else { + Ok(AipSessionStatus::Stale) + }; + } + // Get AIP hostname from config let aip_hostname = match &web_context.config.oauth_backend { OAuthBackendConfig::AIP { hostname, .. } => hostname, @@ -120,6 +209,9 @@ pub(crate) async fn require_valid_aip_session( WebError::InternalError })?; + // Cache the result + cache_aip_session_status(web_context, access_token, is_ready).await; + if is_ready { Ok(AipSessionStatus::Ready) } else { diff --git a/src/http/errors/lfg_error.rs b/src/http/errors/lfg_error.rs new file mode 100644 index 0000000..575cb60 --- /dev/null +++ b/src/http/errors/lfg_error.rs @@ -0,0 +1,79 @@ +use thiserror::Error; + +/// Represents errors that can occur during LFG (Looking For Group) operations. +/// +/// These errors are typically triggered during validation of user-submitted +/// LFG creation forms or during LFG record operations. +#[derive(Debug, Error)] +pub(crate) enum LfgError { + /// Error when the location is not provided. + /// + /// This error occurs when a user attempts to create an LFG record without + /// selecting a location on the map. + #[error("error-smokesignal-lfg-1 Location not set")] + LocationNotSet, + + /// Error when the coordinates are invalid. + /// + /// This error occurs when the provided latitude or longitude + /// values are not valid numbers or are out of range. + #[error("error-smokesignal-lfg-2 Invalid coordinates: {0}")] + InvalidCoordinates(String), + + /// Error when no tags are provided. + /// + /// This error occurs when a user attempts to create an LFG record without + /// specifying at least one interest tag. + #[error("error-smokesignal-lfg-3 Tags required (at least one)")] + TagsRequired, + + /// Error when too many tags are provided. + /// + /// This error occurs when a user attempts to create an LFG record with + /// more than the maximum allowed number of tags (10). + #[error("error-smokesignal-lfg-4 Too many tags (maximum 10)")] + TooManyTags, + + /// Error when an invalid duration is specified. + /// + /// This error occurs when the provided duration value is not one of + /// the allowed options (6, 12, 24, 48, or 72 hours). + #[error("error-smokesignal-lfg-5 Invalid duration")] + InvalidDuration, + + /// Error when the PDS record creation fails. + /// + /// This error occurs when the AT Protocol server returns an error + /// during LFG record creation. + #[error("error-smokesignal-lfg-6 Failed to create PDS record: {message}")] + PdsRecordCreationFailed { message: String }, + + /// Error when no active LFG record is found. + /// + /// This error occurs when attempting to perform operations that + /// require an active LFG record (e.g., deactivation, viewing matches). + #[error("error-smokesignal-lfg-7 No active LFG record found")] + NoActiveRecord, + + /// Error when user already has an active LFG record. + /// + /// This error occurs when a user attempts to create a new LFG record + /// while they already have an active one. Users must deactivate their + /// existing record before creating a new one. + #[error("error-smokesignal-lfg-8 Active LFG record already exists")] + ActiveRecordExists, + + /// Error when deactivation fails. + /// + /// This error occurs when the attempt to deactivate an LFG record + /// fails due to a server or network error. + #[error("error-smokesignal-lfg-9 Failed to deactivate LFG record: {message}")] + DeactivationFailed { message: String }, + + /// Error when a tag is invalid. + /// + /// This error occurs when a provided tag is empty or exceeds + /// the maximum allowed length. + #[error("error-smokesignal-lfg-10 Invalid tag: {0}")] + InvalidTag(String), +} diff --git a/src/http/errors/mod.rs b/src/http/errors/mod.rs index 2432034..eb4e561 100644 --- a/src/http/errors/mod.rs +++ b/src/http/errors/mod.rs @@ -8,6 +8,7 @@ pub mod create_rsvp_errors; pub mod delete_event_errors; pub mod event_view_errors; pub mod import_error; +pub mod lfg_error; pub mod login_error; pub mod profile_import_error; pub mod middleware_errors; @@ -21,6 +22,7 @@ pub(crate) use create_rsvp_errors::CreateRsvpError; pub(crate) use delete_event_errors::DeleteEventError; pub(crate) use event_view_errors::EventViewError; pub(crate) use import_error::ImportError; +pub(crate) use lfg_error::LfgError; pub(crate) use login_error::LoginError; pub(crate) use profile_import_error::ProfileImportError; pub(crate) use middleware_errors::WebSessionError; diff --git a/src/http/errors/web_error.rs b/src/http/errors/web_error.rs index 134219a..51213a5 100644 --- a/src/http/errors/web_error.rs +++ b/src/http/errors/web_error.rs @@ -18,6 +18,7 @@ use super::common_error::CommonError; use super::create_event_errors::CreateEventError; use super::event_view_errors::EventViewError; use super::import_error::ImportError; +use super::lfg_error::LfgError; use super::login_error::LoginError; use super::middleware_errors::MiddlewareAuthError; use super::url_error::UrlError; @@ -159,6 +160,13 @@ pub(crate) enum WebError { #[error(transparent)] BlobError(#[from] BlobError), + /// Looking For Group (LFG) errors. + /// + /// This error occurs when there are issues with LFG operations, + /// such as creating, viewing, or deactivating LFG records. + #[error(transparent)] + LfgError(#[from] LfgError), + /// The AIP session has expired and the user must re-authenticate. /// /// This error occurs when an AT Protocol operation is attempted with a stale diff --git a/src/http/h3_utils.rs b/src/http/h3_utils.rs new file mode 100644 index 0000000..a19ecf8 --- /dev/null +++ b/src/http/h3_utils.rs @@ -0,0 +1,233 @@ +//! H3 geospatial indexing utilities. +//! +//! This module provides helper functions for working with H3 hexagonal +//! hierarchical spatial indexes. + +use h3o::{CellIndex, LatLng, Resolution}; + +/// Default H3 resolution for LFG location selection (precision 6 = ~36km edge) +pub const DEFAULT_RESOLUTION: Resolution = Resolution::Six; + +/// Convert lat/lon to H3 cell index at the default resolution (6). +/// +/// # Arguments +/// * `lat` - Latitude in degrees (-90 to 90) +/// * `lon` - Longitude in degrees (-180 to 180) +/// +/// # Returns +/// The H3 cell index as a string, or an error message. +pub fn lat_lon_to_h3(lat: f64, lon: f64) -> Result { + lat_lon_to_h3_with_resolution(lat, lon, DEFAULT_RESOLUTION) +} + +/// Convert lat/lon to H3 cell index at a specific resolution. +/// +/// # Arguments +/// * `lat` - Latitude in degrees (-90 to 90) +/// * `lon` - Longitude in degrees (-180 to 180) +/// * `resolution` - H3 resolution (0-15) +/// +/// # Returns +/// The H3 cell index as a string, or an error message. +pub fn lat_lon_to_h3_with_resolution( + lat: f64, + lon: f64, + resolution: Resolution, +) -> Result { + let coord = + LatLng::new(lat, lon).map_err(|e| format!("Invalid coordinates: {}", e))?; + let cell = coord.to_cell(resolution); + Ok(cell.to_string()) +} + +/// Validate an H3 index string and parse it into a CellIndex. +/// +/// # Arguments +/// * `h3_str` - The H3 index as a hexadecimal string +/// +/// # Returns +/// The parsed CellIndex, or an error message. +pub fn validate_h3_index(h3_str: &str) -> Result { + h3_str + .parse::() + .map_err(|e| format!("Invalid H3 index: {}", e)) +} + +/// Get the center coordinates of an H3 cell. +/// +/// # Arguments +/// * `h3_str` - The H3 index as a hexadecimal string +/// +/// # Returns +/// A tuple of (latitude, longitude) for the cell center, or an error message. +pub fn h3_to_lat_lon(h3_str: &str) -> Result<(f64, f64), String> { + let cell = validate_h3_index(h3_str)?; + let center = LatLng::from(cell); + Ok((center.lat(), center.lng())) +} + +/// Get neighboring H3 cells within k rings. +/// +/// # Arguments +/// * `h3_str` - The H3 index as a hexadecimal string +/// * `k` - The number of rings to include (0 = just the cell, 1 = cell + immediate neighbors) +/// +/// # Returns +/// A vector of H3 cell indexes as strings, or an error message. +pub fn h3_neighbors(h3_str: &str, k: u32) -> Result, String> { + let cell = validate_h3_index(h3_str)?; + let neighbors: Vec = cell + .grid_disk::>(k) + .into_iter() + .map(|c| c.to_string()) + .collect(); + Ok(neighbors) +} + +/// Get the boundary vertices of an H3 cell for map display. +/// +/// # Arguments +/// * `h3_str` - The H3 index as a hexadecimal string +/// +/// # Returns +/// A vector of (latitude, longitude) tuples forming the cell boundary, or an error message. +pub fn h3_boundary(h3_str: &str) -> Result, String> { + let cell = validate_h3_index(h3_str)?; + let boundary = cell.boundary(); + Ok(boundary.iter().map(|v| (v.lat(), v.lng())).collect()) +} + +/// Get the resolution of an H3 cell. +/// +/// # Arguments +/// * `h3_str` - The H3 index as a hexadecimal string +/// +/// # Returns +/// The resolution (0-15), or an error message. +pub fn h3_resolution(h3_str: &str) -> Result { + let cell = validate_h3_index(h3_str)?; + Ok(cell.resolution() as u8) +} + +/// Calculate the approximate area of an H3 cell in square kilometers. +/// +/// # Arguments +/// * `h3_str` - The H3 index as a hexadecimal string +/// +/// # Returns +/// The area in square kilometers, or an error message. +pub fn h3_area_km2(h3_str: &str) -> Result { + let cell = validate_h3_index(h3_str)?; + Ok(cell.area_km2()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_lat_lon_to_h3() { + // New York City coordinates + let result = lat_lon_to_h3(40.7128, -74.0060); + assert!(result.is_ok()); + let h3_index = result.unwrap(); + assert!(!h3_index.is_empty()); + // H3 indexes are 15-character hex strings + assert_eq!(h3_index.len(), 15); + } + + #[test] + fn test_lat_lon_to_h3_invalid() { + // Invalid latitude + let result = lat_lon_to_h3(100.0, 0.0); + assert!(result.is_err()); + + // Invalid longitude + let result = lat_lon_to_h3(0.0, 200.0); + assert!(result.is_err()); + } + + #[test] + fn test_h3_to_lat_lon() { + // Convert NYC to H3 and back + let h3_index = lat_lon_to_h3(40.7128, -74.0060).unwrap(); + let (lat, lon) = h3_to_lat_lon(&h3_index).unwrap(); + + // Should be close to original (within cell) + assert!((lat - 40.7128).abs() < 1.0); + assert!((lon - (-74.0060)).abs() < 1.0); + } + + #[test] + fn test_validate_h3_index() { + let h3_index = lat_lon_to_h3(40.7128, -74.0060).unwrap(); + let result = validate_h3_index(&h3_index); + assert!(result.is_ok()); + } + + #[test] + fn test_validate_h3_index_invalid() { + let result = validate_h3_index("invalid"); + assert!(result.is_err()); + } + + #[test] + fn test_h3_neighbors() { + let h3_index = lat_lon_to_h3(40.7128, -74.0060).unwrap(); + + // k=0 should return just the cell itself + let neighbors_0 = h3_neighbors(&h3_index, 0).unwrap(); + assert_eq!(neighbors_0.len(), 1); + assert_eq!(neighbors_0[0], h3_index); + + // k=1 should return the cell plus 6 neighbors (7 total) + let neighbors_1 = h3_neighbors(&h3_index, 1).unwrap(); + assert_eq!(neighbors_1.len(), 7); + assert!(neighbors_1.contains(&h3_index)); + } + + #[test] + fn test_h3_boundary() { + let h3_index = lat_lon_to_h3(40.7128, -74.0060).unwrap(); + let boundary = h3_boundary(&h3_index).unwrap(); + + // H3 cells are hexagons, so they have 6 vertices + assert_eq!(boundary.len(), 6); + + // All vertices should be valid coordinates + for (lat, lon) in &boundary { + assert!((-90.0..=90.0).contains(lat)); + assert!((-180.0..=180.0).contains(lon)); + } + } + + #[test] + fn test_h3_resolution() { + let h3_index = lat_lon_to_h3(40.7128, -74.0060).unwrap(); + let resolution = h3_resolution(&h3_index).unwrap(); + assert_eq!(resolution, 6); // Default resolution + } + + #[test] + fn test_h3_area_km2() { + let h3_index = lat_lon_to_h3(40.7128, -74.0060).unwrap(); + let area = h3_area_km2(&h3_index).unwrap(); + + // Resolution 6 cells are approximately 36 km^2 + assert!(area > 30.0 && area < 50.0); + } + + #[test] + fn test_different_resolutions() { + // Test resolution 5 (larger cells) + let h3_res5 = lat_lon_to_h3_with_resolution(40.7128, -74.0060, Resolution::Five).unwrap(); + let area_5 = h3_area_km2(&h3_res5).unwrap(); + + // Test resolution 7 (smaller cells) + let h3_res7 = lat_lon_to_h3_with_resolution(40.7128, -74.0060, Resolution::Seven).unwrap(); + let area_7 = h3_area_km2(&h3_res7).unwrap(); + + // Resolution 5 cells should be larger than resolution 7 + assert!(area_5 > area_7); + } +} diff --git a/src/http/handle_edit_event.rs b/src/http/handle_edit_event.rs index aef23a0..4f15771 100644 --- a/src/http/handle_edit_event.rs +++ b/src/http/handle_edit_event.rs @@ -6,6 +6,7 @@ use serde_json::json; use crate::atproto::auth::{ create_dpop_auth_from_aip_session, create_dpop_auth_from_oauth_session, }; +use crate::search_index::SearchIndexManager; use crate::http::auth_utils::{require_valid_aip_session, AipSessionStatus}; use crate::atproto::utils::{location_from_address, location_from_geo}; use crate::http::context::UserRequestContext; @@ -435,6 +436,43 @@ pub(crate) async fn handle_edit_event_json( ).into_response()); } + // Re-index the event in OpenSearch to update locations_geo and other fields + if let Some(endpoint) = &ctx.web_context.config.opensearch_endpoint { + if let Ok(manager) = SearchIndexManager::new(endpoint) { + // Fetch the updated event from the database + match event_get(&ctx.web_context.pool, &put_record_response.uri).await { + Ok(updated_event) => { + if let Err(err) = manager + .index_event( + &ctx.web_context.pool, + ctx.web_context.identity_resolver.clone(), + &updated_event, + ) + .await + { + tracing::warn!( + ?err, + aturi = %put_record_response.uri, + "Failed to re-index event in OpenSearch after edit" + ); + } else { + tracing::info!( + aturi = %put_record_response.uri, + "Successfully re-indexed event in OpenSearch after edit" + ); + } + } + Err(err) => { + tracing::warn!( + ?err, + aturi = %put_record_response.uri, + "Failed to fetch updated event for OpenSearch indexing" + ); + } + } + } + } + // Download and store header image from PDS if one was uploaded if let Some(ref header_cid) = request.header_cid && let Err(err) = store_event_header_from_pds( diff --git a/src/http/handle_geo_aggregation.rs b/src/http/handle_geo_aggregation.rs index 6b4b958..3ddd976 100644 --- a/src/http/handle_geo_aggregation.rs +++ b/src/http/handle_geo_aggregation.rs @@ -9,9 +9,9 @@ use crate::http::errors::WebError; use crate::search_index::{GeoCenter, GeoHexBucket, SearchIndexManager}; /// H3 precision level (5 = ~8.5km edge length) -const PRECISION: u8 = 6; +const PRECISION: u8 = 7; /// Search radius in miles -const DISTANCE_MILES: f64 = 300.0; +const DISTANCE_MILES: f64 = 60.0; #[derive(Debug, Deserialize)] pub(crate) struct GeoAggregationParams { diff --git a/src/http/handle_index.rs b/src/http/handle_index.rs index 2c54504..09e11ae 100644 --- a/src/http/handle_index.rs +++ b/src/http/handle_index.rs @@ -414,10 +414,14 @@ pub(crate) async fn handle_index( .unwrap_or(crate::stats::NetworkStats { event_count: 0, rsvp_count: 0, + lfg_identities_count: 0, + lfg_locations_count: 0, }); let event_count_formatted = crate::stats::NetworkStats::format_number(stats.event_count); let rsvp_count_formatted = crate::stats::NetworkStats::format_number(stats.rsvp_count); + let lfg_identities_count_formatted = crate::stats::NetworkStats::format_number(stats.lfg_identities_count); + let lfg_locations_count_formatted = crate::stats::NetworkStats::format_number(stats.lfg_locations_count); Ok(( http::StatusCode::OK, @@ -433,6 +437,8 @@ pub(crate) async fn handle_index( pagination => pagination_view, event_count => event_count_formatted, rsvp_count => rsvp_count_formatted, + lfg_identities_count => lfg_identities_count_formatted, + lfg_locations_count => lfg_locations_count_formatted, }, ), ) diff --git a/src/http/handle_lfg.rs b/src/http/handle_lfg.rs new file mode 100644 index 0000000..45fc706 --- /dev/null +++ b/src/http/handle_lfg.rs @@ -0,0 +1,855 @@ +//! HTTP handlers for the Looking For Group (LFG) feature. +//! +//! This module provides handlers for creating, viewing, and managing LFG records +//! that allow users to find activity partners in their geographic area. + +use atproto_client::com::atproto::repo::{ + CreateRecordRequest, CreateRecordResponse, PutRecordRequest, PutRecordResponse, create_record, + put_record, +}; +use atproto_record::lexicon::community::lexicon::location::{Hthree, LocationOrRef, TypedHthree}; +use axum::Json; +use axum::extract::State; +use axum::response::IntoResponse; +use axum_extra::extract::Cached; +use axum_extra::extract::Query; +use axum_htmx::{HxBoosted, HxRequest}; +use axum_template::RenderHtml; +use chrono::{Duration, Utc}; +use h3o::{LatLng, Resolution}; +use minijinja::context as template_context; +use serde::{Deserialize, Serialize}; + +use std::collections::HashMap; + +use crate::atproto::auth::{ + create_dpop_auth_from_aip_session, create_dpop_auth_from_oauth_session, +}; +use crate::atproto::lexicon::lfg::{Lfg, NSID}; +use crate::config::OAuthBackendConfig; +use crate::http::auth_utils::{AipSessionStatus, require_valid_aip_session}; +use crate::http::context::WebContext; +use crate::http::errors::{CommonError, LfgError, WebError}; +use crate::http::event_view::EventView; +use crate::http::lfg_form::{ALLOWED_DURATIONS, DEFAULT_DURATION_HOURS, MAX_TAGS, MAX_TAG_LENGTH}; +use crate::http::middleware_auth::Auth; +use crate::http::middleware_i18n::Language; +use crate::search_index::{GeoCenter, IndexedEvent, IndexedLfgProfile, SearchIndexManager}; +use crate::select_template; +use crate::storage::atproto_record::atproto_record_upsert; +use crate::storage::event::event_get; +use crate::storage::identity_profile::{handle_for_did, handles_by_did}; +use crate::storage::lfg::{lfg_get_active_by_did, lfg_get_all_by_did}; +use crate::storage::profile::profile_get_by_did; +use crate::storage::StoragePool; + +// ============================================================================ +// Request/Response Types +// ============================================================================ + +/// Request body for creating an LFG record. +#[derive(Debug, Deserialize)] +pub struct CreateLfgRequest { + /// Latitude of the location + pub latitude: f64, + /// Longitude of the location + pub longitude: f64, + /// Interest tags (1-10 items) + pub tags: Vec, + /// Duration in hours (6, 12, 24, 48, or 72) + pub duration_hours: u32, +} + +/// Response for successful LFG creation. +#[derive(Debug, Serialize)] +pub struct CreateLfgResponse { + /// AT-URI of the created record + pub aturi: String, + /// CID of the record + pub cid: String, +} + +/// Error response for LFG operations. +#[derive(Debug, Serialize)] +pub struct LfgErrorResponse { + /// Error code + pub error: String, + /// Human-readable error message + pub message: String, +} + +/// Query parameters for tag autocomplete +#[derive(Debug, Deserialize)] +pub struct TagAutocompleteQuery { + /// Search query (prefix match) + #[serde(default)] + pub q: String, + /// Maximum number of results + #[serde(default = "default_limit")] + pub limit: u32, +} + +fn default_limit() -> u32 { + 10 +} + +/// Tag suggestion in autocomplete response +#[derive(Debug, Serialize)] +pub struct TagSuggestion { + /// Tag name + pub name: String, + /// Usage count + pub count: i64, + /// Source: "history" (user's past tags) or "popular" (global) + pub source: String, +} + +/// Response for tag autocomplete endpoint +#[derive(Debug, Serialize)] +pub struct TagAutocompleteResponse { + pub tags: Vec, +} + +/// Query parameters for geo aggregation +#[derive(Debug, Deserialize)] +pub struct GeoAggregationQuery { + /// Center latitude (optional, uses LFG location if available) + pub lat: Option, + /// Center longitude (optional, uses LFG location if available) + pub lon: Option, + /// H3 precision (default 6) + #[serde(default = "default_precision")] + pub precision: u8, +} + +fn default_precision() -> u8 { + 6 +} + +/// Bucket in geo aggregation response +#[derive(Debug, Serialize)] +pub struct GeoHexBucket { + /// H3 cell index + pub key: String, + /// Total count (events + people) + pub doc_count: u64, + /// Event count + pub event_count: u64, + /// People count + pub people_count: u64, +} + +/// Response for geo aggregation endpoint +#[derive(Debug, Serialize)] +pub struct GeoAggregationResponse { + pub buckets: Vec, + pub lat: f64, + pub lon: f64, +} + +// ============================================================================ +// Helper Functions +// ============================================================================ + +/// Helper to get popular tags from OpenSearch with graceful fallback. +/// +/// Returns an empty vector if OpenSearch is not configured or unavailable. +async fn get_popular_tags(opensearch_endpoint: Option<&str>, limit: u32) -> Vec<(String, i64)> { + let Some(endpoint) = opensearch_endpoint else { + return vec![]; + }; + + let Ok(manager) = SearchIndexManager::new(endpoint) else { + return vec![]; + }; + + manager + .get_popular_lfg_tags(limit) + .await + .unwrap_or_default() +} + + +/// Serializable LFG profile for template display +#[derive(Debug, Serialize)] +struct TemplateProfile { + did: String, + handle: Option, + display_name: Option, + tags: Vec, +} + +impl TemplateProfile { + /// Create from IndexedLfgProfile with optional profile enrichment + fn from_indexed(p: IndexedLfgProfile) -> Self { + Self { + did: p.did, + handle: None, + display_name: None, + tags: p.tags, + } + } +} + +/// Serializable geo bucket for template display +#[derive(Debug, Serialize)] +struct TemplateBucket { + key: String, + count: u64, +} + +/// Enrich LFG profiles with display_name and handle from the database. +async fn enrich_profiles(pool: &StoragePool, profiles: Vec) -> Vec { + let mut enriched = Vec::with_capacity(profiles.len()); + + for p in profiles { + let did = p.did.clone(); + let mut template_profile = TemplateProfile::from_indexed(p); + + // Try to get AT Protocol profile for display_name + if let Ok(Some(profile)) = profile_get_by_did(pool, &did).await { + if !profile.display_name.is_empty() { + template_profile.display_name = Some(profile.display_name); + } + } + + // Try to get identity profile for handle + if let Ok(identity) = handle_for_did(pool, &did).await { + template_profile.handle = Some(identity.handle); + } + + enriched.push(template_profile); + } + + enriched +} + +/// Query OpenSearch for nearby events, profiles, and geo aggregations. +/// +/// Returns a tuple of (indexed_events, profiles, event_buckets, profile_buckets). +/// Indexed events need to be further enriched by fetching from the database. +async fn query_nearby_activity( + pool: &StoragePool, + opensearch_endpoint: Option<&str>, + lat: f64, + lon: f64, + distance_miles: f64, + tags: &[String], +) -> ( + Vec, + Vec, + Vec, + Vec, +) { + let Some(endpoint) = opensearch_endpoint else { + return (vec![], vec![], vec![], vec![]); + }; + + let Ok(manager) = SearchIndexManager::new(endpoint) else { + return (vec![], vec![], vec![], vec![]); + }; + + let center = GeoCenter { + lat, + lon, + distance_miles, + }; + + // Run queries in parallel + let (events_result, profiles_result, event_agg_result, profile_agg_result) = tokio::join!( + manager.search_nearby_upcoming_events(lat, lon, distance_miles, 20), + manager.search_nearby_lfg_profiles(lat, lon, distance_miles, tags, None, 20), + manager.get_event_geo_aggregation(7, Some(center.clone()), true), + manager.get_lfg_profile_geo_aggregation(7, Some(center)), + ); + + let indexed_events = events_result.unwrap_or_default(); + + // Enrich profiles with display_name and handle + let indexed_profiles = profiles_result.unwrap_or_default(); + let profiles = enrich_profiles(pool, indexed_profiles).await; + + let event_buckets: Vec = event_agg_result + .unwrap_or_default() + .into_iter() + .map(|b| TemplateBucket { + key: b.key, + count: b.doc_count, + }) + .collect(); + + let profile_buckets: Vec = profile_agg_result + .unwrap_or_default() + .into_iter() + .map(|b| TemplateBucket { + key: b.key, + count: b.doc_count, + }) + .collect(); + + (indexed_events, profiles, event_buckets, profile_buckets) +} + +/// Validate a CreateLfgRequest and return an error if invalid. +fn validate_create_lfg_request(request: &CreateLfgRequest) -> Result<(), LfgError> { + // Validate coordinates + if !(-90.0..=90.0).contains(&request.latitude) { + return Err(LfgError::InvalidCoordinates( + "latitude out of range".to_string(), + )); + } + if !(-180.0..=180.0).contains(&request.longitude) { + return Err(LfgError::InvalidCoordinates( + "longitude out of range".to_string(), + )); + } + + // Validate tags + if request.tags.is_empty() { + return Err(LfgError::TagsRequired); + } + if request.tags.len() > MAX_TAGS { + return Err(LfgError::TooManyTags); + } + for tag in &request.tags { + let trimmed = tag.trim(); + if trimmed.is_empty() { + return Err(LfgError::InvalidTag("empty tag".to_string())); + } + if trimmed.len() > MAX_TAG_LENGTH { + return Err(LfgError::InvalidTag(format!( + "tag exceeds {} characters", + MAX_TAG_LENGTH + ))); + } + } + + // Validate duration + if !ALLOWED_DURATIONS.contains(&request.duration_hours) { + return Err(LfgError::InvalidDuration); + } + + Ok(()) +} + +// ============================================================================ +// GET Handler - Display Form or Matches +// ============================================================================ + +/// GET /lfg - Display the LFG form or matches view +/// +/// If the user has no active LFG record, shows the creation form. +/// If the user has an active LFG record, shows the matches view. +pub(crate) async fn handle_lfg_get( + State(web_context): State, + Language(language): Language, + Cached(auth): Cached, + HxRequest(hx_request): HxRequest, + HxBoosted(hx_boosted): HxBoosted, +) -> Result { + let current_handle = auth.require("/lfg")?; + + let is_development = cfg!(debug_assertions); + + let default_context = template_context! { + current_handle => current_handle.clone(), + language => language.to_string(), + canonical_url => format!("https://{}/lfg", web_context.config.external_base), + is_development, + allowed_durations => ALLOWED_DURATIONS, + default_duration => DEFAULT_DURATION_HOURS, + }; + + // Check for existing active LFG record + let active_lfg = lfg_get_active_by_did(&web_context.pool, ¤t_handle.did).await?; + + match active_lfg { + None => { + // No active LFG - show creation form + let popular_tags = + get_popular_tags(web_context.config.opensearch_endpoint.as_deref(), 20).await; + + Ok(RenderHtml( + select_template!("lfg_form", hx_boosted, hx_request, language), + web_context.engine.clone(), + template_context! { ..default_context, ..template_context! { + popular_tags, + }}, + ) + .into_response()) + } + Some(lfg_record) => { + // User has an active LFG - show matches view + let lfg: Lfg = serde_json::from_value(lfg_record.record.0.clone()) + .map_err(|_| LfgError::NoActiveRecord)?; + + // Extract coordinates and H3 cell from the LFG record + let (lat, lon) = lfg.get_coordinates().ok_or(LfgError::LocationNotSet)?; + let h3_cell = lfg.get_h3_cell().map(|c| c.to_string()); + + // Query OpenSearch for nearby events and profiles + let search_radius_miles = 100.0; // Search within 100 miles + let (indexed_events, matching_profiles, event_buckets, profile_buckets) = + query_nearby_activity( + &web_context.pool, + web_context.config.opensearch_endpoint.as_deref(), + lat, + lon, + search_radius_miles, + &lfg.tags, + ) + .await; + + // Fetch full events from database + let mut db_events = vec![]; + for indexed_event in &indexed_events { + match event_get(&web_context.pool, &indexed_event.aturi).await { + Ok(event) => db_events.push(event), + Err(err) => { + tracing::warn!("Failed to fetch event {}: {}", indexed_event.aturi, err); + } + } + } + + // Get organizer handles + let event_dids: Vec = db_events.iter().map(|e| e.did.clone()).collect(); + let organizer_handles = handles_by_did(&web_context.pool, event_dids) + .await + .unwrap_or_else(|_| HashMap::new()); + + // Build EventViews + let facet_limits = crate::facets::FacetLimits { + mentions_max: web_context.config.facets_mentions_max, + tags_max: web_context.config.facets_tags_max, + links_max: web_context.config.facets_links_max, + max: web_context.config.facets_max, + }; + + let mut events: Vec = db_events + .iter() + .filter_map(|event| { + let organizer = organizer_handles.get(&event.did); + EventView::try_from(( + auth.profile(), + organizer, + event, + &facet_limits, + )) + .ok() + }) + .collect(); + + // Hydrate RSVP counts + if let Err(err) = crate::http::event_view::hydrate_event_rsvp_counts( + &web_context.pool, + &mut events, + ) + .await + { + tracing::warn!("Failed to hydrate event RSVP counts: {}", err); + } + + // Extract user's tags for highlighting matches in the template + let user_tags: Vec = lfg.tags.iter().map(|t| t.to_lowercase()).collect(); + + Ok(RenderHtml( + select_template!("lfg_matches", hx_boosted, hx_request, language), + web_context.engine.clone(), + template_context! { ..default_context, ..template_context! { + lfg_record => serde_json::to_value(&lfg).ok(), + lfg_aturi => lfg_record.aturi, + latitude => lat, + longitude => lon, + h3_cell, + events, + matching_profiles, + event_buckets, + profile_buckets, + user_tags, + }}, + ) + .into_response()) + } + } +} + +// ============================================================================ +// POST Handler - Create LFG Record (JSON) +// ============================================================================ + +/// POST /lfg - Create a new LFG record +/// +/// Accepts a JSON body with location, tags, and duration. +/// Returns JSON with the created record's AT-URI and CID. +pub(crate) async fn handle_lfg_post( + State(web_context): State, + Cached(auth): Cached, + Json(request): Json, +) -> Result, WebError> { + let current_handle = auth.require("/lfg")?; + + // Check AIP session validity before attempting AT Protocol operation + if let AipSessionStatus::Stale = require_valid_aip_session(&web_context, &auth).await? { + return Err(WebError::SessionStale); + } + + // Validate the request + validate_create_lfg_request(&request)?; + + // Check if user already has an active LFG record + let active_lfg = lfg_get_active_by_did(&web_context.pool, ¤t_handle.did).await?; + if active_lfg.is_some() { + return Err(LfgError::ActiveRecordExists.into()); + } + + // Normalize tags (trim, remove duplicates case-insensitively, preserve original case) + let tags: Vec = { + let mut seen = std::collections::HashSet::new(); + request + .tags + .iter() + .map(|t| t.trim().to_string()) + .filter(|t| !t.is_empty() && seen.insert(t.to_lowercase())) + .collect() + }; + + // Convert lat/lng to H3 cell at resolution 7 + let lat_lng = + LatLng::new(request.latitude, request.longitude).map_err(|_| LfgError::LocationNotSet)?; + let cell = lat_lng.to_cell(Resolution::Seven); + + let location = LocationOrRef::InlineHthree(TypedHthree::new(Hthree { + value: cell.to_string(), + name: None, + })); + + // Create the LFG record + let now = Utc::now(); + let ends_at = now + Duration::hours(request.duration_hours as i64); + + let lfg_record = Lfg { + location, + tags, + starts_at: now, + ends_at, + created_at: now, + active: true, + }; + + // Create DPoP auth based on OAuth backend type + let dpop_auth = match (&auth, &web_context.config.oauth_backend) { + (Auth::Pds { session, .. }, OAuthBackendConfig::ATProtocol { .. }) => { + create_dpop_auth_from_oauth_session(session)? + } + (Auth::Aip { access_token, .. }, OAuthBackendConfig::AIP { hostname, .. }) => { + create_dpop_auth_from_aip_session(&web_context.http_client, hostname, access_token) + .await? + } + _ => return Err(CommonError::NotAuthorized.into()), + }; + + let create_request = CreateRecordRequest { + repo: current_handle.did.clone(), + collection: NSID.to_string(), + validate: false, + record_key: None, + record: lfg_record.clone(), + swap_commit: None, + }; + + let create_result = create_record( + &web_context.http_client, + &atproto_client::client::Auth::DPoP(dpop_auth), + ¤t_handle.pds, + create_request, + ) + .await; + + let (aturi, cid) = match create_result { + Ok(CreateRecordResponse::StrongRef { uri, cid, .. }) => (uri, cid), + Ok(CreateRecordResponse::Error(err)) => { + return Err(LfgError::PdsRecordCreationFailed { + message: err.error_message(), + } + .into()); + } + Err(err) => { + return Err(LfgError::PdsRecordCreationFailed { + message: err.to_string(), + } + .into()); + } + }; + + // Store in local database + let record_json = serde_json::to_value(&lfg_record).map_err(|e| { + LfgError::PdsRecordCreationFailed { + message: e.to_string(), + } + })?; + + atproto_record_upsert( + &web_context.pool, + &aturi, + ¤t_handle.did, + &cid, + NSID, + &record_json, + ) + .await?; + + // Index to OpenSearch + if let Some(endpoint) = &web_context.config.opensearch_endpoint { + if let Ok(manager) = SearchIndexManager::new(endpoint) { + if let Some((lat, lon)) = lfg_record.get_coordinates() { + if let Err(e) = manager + .index_lfg_profile( + &aturi, + ¤t_handle.did, + lat, + lon, + &lfg_record.tags, + &lfg_record.starts_at, + &lfg_record.ends_at, + &lfg_record.created_at, + true, // active = true + ) + .await + { + tracing::warn!("Failed to index LFG profile to search index: {}", e); + } + } + } + } + + Ok(Json(CreateLfgResponse { + aturi, + cid: cid.to_string(), + })) +} + +// ============================================================================ +// POST Handler - Deactivate LFG Record +// ============================================================================ + +/// POST /lfg/deactivate - Deactivate the user's active LFG record +pub(crate) async fn handle_lfg_deactivate( + State(web_context): State, + Cached(auth): Cached, +) -> Result { + let current_handle = auth.require("/lfg/deactivate")?; + + // Check AIP session validity + if let AipSessionStatus::Stale = require_valid_aip_session(&web_context, &auth).await? { + return Err(WebError::SessionStale); + } + + // Get the active LFG record + let active_lfg = lfg_get_active_by_did(&web_context.pool, ¤t_handle.did) + .await? + .ok_or(LfgError::NoActiveRecord)?; + + // Parse the existing record and create updated version with active=false + let mut lfg: Lfg = serde_json::from_value(active_lfg.record.0.clone()) + .map_err(|_| LfgError::NoActiveRecord)?; + lfg.active = false; + + // Create DPoP auth + let dpop_auth = match (&auth, &web_context.config.oauth_backend) { + (Auth::Pds { session, .. }, OAuthBackendConfig::ATProtocol { .. }) => { + create_dpop_auth_from_oauth_session(session)? + } + (Auth::Aip { access_token, .. }, OAuthBackendConfig::AIP { hostname, .. }) => { + create_dpop_auth_from_aip_session(&web_context.http_client, hostname, access_token) + .await? + } + _ => return Err(CommonError::NotAuthorized.into()), + }; + + // Extract rkey from AT-URI + let rkey = active_lfg + .aturi + .rsplit('/') + .next() + .ok_or(LfgError::NoActiveRecord)?; + + // Update the record on PDS with active=false + let put_request = PutRecordRequest { + repo: current_handle.did.clone(), + collection: NSID.to_string(), + record_key: rkey.to_string(), + validate: false, + record: lfg.clone(), + swap_record: None, + swap_commit: None, + }; + + let put_result = put_record( + &web_context.http_client, + &atproto_client::client::Auth::DPoP(dpop_auth), + ¤t_handle.pds, + put_request, + ) + .await; + + let (aturi, cid) = match put_result { + Ok(PutRecordResponse::StrongRef { uri, cid, .. }) => (uri, cid), + Ok(PutRecordResponse::Error(err)) => { + return Err(LfgError::DeactivationFailed { + message: err.error_message(), + } + .into()); + } + Err(err) => { + return Err(LfgError::DeactivationFailed { + message: err.to_string(), + } + .into()); + } + }; + + // Update local database with deactivated record + let record_json = serde_json::to_value(&lfg).map_err(|e| LfgError::DeactivationFailed { + message: e.to_string(), + })?; + + atproto_record_upsert( + &web_context.pool, + &aturi, + ¤t_handle.did, + &cid, + NSID, + &record_json, + ) + .await?; + + // Update OpenSearch index with active=false + if let Some(endpoint) = &web_context.config.opensearch_endpoint { + if let Ok(manager) = SearchIndexManager::new(endpoint) { + if let Some((lat, lon)) = lfg.get_coordinates() { + if let Err(e) = manager + .index_lfg_profile( + &aturi, + ¤t_handle.did, + lat, + lon, + &lfg.tags, + &lfg.starts_at, + &lfg.ends_at, + &lfg.created_at, + false, // active = false + ) + .await + { + tracing::warn!("Failed to update LFG profile in search index: {}", e); + } + } + } + } + + // Redirect to LFG page (will show form since no active record) + Ok(axum::response::Redirect::to("/lfg").into_response()) +} + +// ============================================================================ +// API Handlers +// ============================================================================ + +/// GET /api/lfg/tags - Tag autocomplete +pub(crate) async fn handle_lfg_tags_autocomplete( + State(web_context): State, + Cached(auth): Cached, + Query(query): Query, +) -> Result, WebError> { + let current_handle = auth.require("/api/lfg/tags")?; + + let mut suggestions: Vec = Vec::new(); + let query_lower = query.q.to_lowercase(); + + // Get user's historical tags + let user_records = lfg_get_all_by_did(&web_context.pool, ¤t_handle.did, 50) + .await + .unwrap_or_default(); + + let mut user_tag_counts: std::collections::HashMap = + std::collections::HashMap::new(); + + for record in &user_records { + if let Ok(lfg) = serde_json::from_value::(record.record.0.clone()) { + for tag in lfg.tags { + let normalized = tag.to_lowercase(); + *user_tag_counts.entry(normalized).or_insert(0) += 1; + } + } + } + + // Add user's historical tags (matching prefix) + for (tag, count) in &user_tag_counts { + if query_lower.is_empty() || tag.starts_with(&query_lower) { + suggestions.push(TagSuggestion { + name: tag.clone(), + count: *count, + source: "history".to_string(), + }); + } + } + + // Get popular tags globally from OpenSearch + let popular_tags = + get_popular_tags(web_context.config.opensearch_endpoint.as_deref(), 50).await; + + // Add popular tags (matching prefix, excluding already added) + for (tag, count) in popular_tags { + let normalized = tag.to_lowercase(); + if !user_tag_counts.contains_key(&normalized) + && (query_lower.is_empty() || normalized.starts_with(&query_lower)) + { + suggestions.push(TagSuggestion { + name: tag, + count, + source: "popular".to_string(), + }); + } + } + + // Sort: history first, then by count descending + suggestions.sort_by(|a, b| { + let source_order = match (a.source.as_str(), b.source.as_str()) { + ("history", "popular") => std::cmp::Ordering::Less, + ("popular", "history") => std::cmp::Ordering::Greater, + _ => std::cmp::Ordering::Equal, + }; + source_order.then(b.count.cmp(&a.count)) + }); + + // Limit results + suggestions.truncate(query.limit as usize); + + Ok(Json(TagAutocompleteResponse { tags: suggestions })) +} + +/// GET /api/lfg/geo-aggregation - Geo aggregation for heatmap +pub(crate) async fn handle_lfg_geo_aggregation( + State(web_context): State, + Cached(auth): Cached, + Query(query): Query, +) -> Result, WebError> { + let current_handle = auth.require("/api/lfg/geo-aggregation")?; + + // Get the user's active LFG record for location + let active_lfg = lfg_get_active_by_did(&web_context.pool, ¤t_handle.did).await?; + + let (lat, lon) = if let Some(ref lfg_record) = active_lfg { + serde_json::from_value::(lfg_record.record.0.clone()) + .ok() + .and_then(|lfg| lfg.get_coordinates()) + .unwrap_or((query.lat.unwrap_or(0.0), query.lon.unwrap_or(0.0))) + } else { + (query.lat.unwrap_or(0.0), query.lon.unwrap_or(0.0)) + }; + + // TODO: Query OpenSearch for geo aggregation + let buckets: Vec = vec![]; + + Ok(Json(GeoAggregationResponse { buckets, lat, lon })) +} diff --git a/src/http/handle_manage_event.rs b/src/http/handle_manage_event.rs index 3d637d0..5ef6cf3 100644 --- a/src/http/handle_manage_event.rs +++ b/src/http/handle_manage_event.rs @@ -270,6 +270,38 @@ pub(crate) async fn handle_manage_event( }) .collect(); + // Extract all locations (addresses and geo) from the event for the form + let mut event_locations: Vec = Vec::new(); + let mut event_geo_locations: Vec = Vec::new(); + + for location in &community_event.locations { + use atproto_record::lexicon::community::lexicon::location::LocationOrRef; + match location { + LocationOrRef::InlineAddress(typed_address) => { + let addr = &typed_address.inner; + event_locations.push(serde_json::json!({ + "country": addr.country, + "postal_code": addr.postal_code, + "region": addr.region, + "locality": addr.locality, + "street": addr.street, + "name": addr.name + })); + } + LocationOrRef::InlineGeo(typed_geo) => { + let geo = &typed_geo.inner; + event_geo_locations.push(serde_json::json!({ + "latitude": geo.latitude, + "longitude": geo.longitude, + "name": geo.name + })); + } + _ => { + // Skip other location types (refs, etc.) + } + } + } + // Load event data for the details and content tabs let (starts_form, location_form, locations_editable, location_edit_reason) = if active_tab == "details" @@ -505,6 +537,8 @@ pub(crate) async fn handle_manage_event( timezones, default_tz, event_links, + event_locations, + event_geo_locations, locations_editable, location_edit_reason, delete_event_url, @@ -544,6 +578,8 @@ pub(crate) async fn handle_manage_event( starts_form, location_form, event_links, + event_locations, + event_geo_locations, popular_countries => popular, other_countries => others, timezones, diff --git a/src/http/handle_oauth_aip_login.rs b/src/http/handle_oauth_aip_login.rs index fa1fb8c..9d7921a 100644 --- a/src/http/handle_oauth_aip_login.rs +++ b/src/http/handle_oauth_aip_login.rs @@ -116,6 +116,7 @@ pub(crate) async fn handle_oauth_aip_login( "repo:community.lexicon.calendar.rsvp", "repo:events.smokesignal.calendar.acceptance", "repo:events.smokesignal.profile", + "repo:events.smokesignal.lfg", "rpc:tools.graze.aip.ready?aud=*", ] .join(" "); diff --git a/src/http/lfg_form.rs b/src/http/lfg_form.rs new file mode 100644 index 0000000..26ab473 --- /dev/null +++ b/src/http/lfg_form.rs @@ -0,0 +1,46 @@ +//! LFG form constants and validation utilities. +//! +//! This module provides constants for the Looking For Group (LFG) feature. + +/// Allowed duration options in hours for LFG records. +pub(crate) const ALLOWED_DURATIONS: [u32; 5] = [6, 12, 24, 48, 72]; + +/// Default duration in hours for new LFG records. +pub(crate) const DEFAULT_DURATION_HOURS: u32 = 48; + +/// Maximum number of tags allowed per LFG record. +pub(crate) const MAX_TAGS: usize = 10; + +/// Maximum length of a single tag. +pub(crate) const MAX_TAG_LENGTH: usize = 64; + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_allowed_durations() { + assert!(ALLOWED_DURATIONS.contains(&6)); + assert!(ALLOWED_DURATIONS.contains(&12)); + assert!(ALLOWED_DURATIONS.contains(&24)); + assert!(ALLOWED_DURATIONS.contains(&48)); + assert!(ALLOWED_DURATIONS.contains(&72)); + assert!(!ALLOWED_DURATIONS.contains(&1)); + assert!(!ALLOWED_DURATIONS.contains(&100)); + } + + #[test] + fn test_default_duration() { + assert_eq!(DEFAULT_DURATION_HOURS, 48); + } + + #[test] + fn test_max_tags() { + assert_eq!(MAX_TAGS, 10); + } + + #[test] + fn test_max_tag_length() { + assert_eq!(MAX_TAG_LENGTH, 64); + } +} diff --git a/src/http/mod.rs b/src/http/mod.rs index b71baa0..a04af8c 100644 --- a/src/http/mod.rs +++ b/src/http/mod.rs @@ -38,6 +38,8 @@ pub mod handle_export_rsvps; pub mod handle_finalize_acceptance; pub mod handle_geo_aggregation; pub mod handle_health; +pub mod handle_lfg; +pub mod h3_utils; pub mod handle_host_meta; pub mod handle_import; pub mod handle_index; @@ -67,6 +69,7 @@ pub mod handle_xrpc_link_attestation; pub mod handle_xrpc_search_events; pub mod handler_mcp; pub mod import_utils; +pub mod lfg_form; pub mod location_edit_status; pub mod macros; pub mod middleware_auth; diff --git a/src/http/server.rs b/src/http/server.rs index ac772a9..9b3596a 100644 --- a/src/http/server.rs +++ b/src/http/server.rs @@ -66,6 +66,10 @@ use crate::http::{ handle_finalize_acceptance::handle_finalize_acceptance, handle_geo_aggregation::handle_geo_aggregation, handle_health::{handle_alive, handle_ready, handle_started}, + handle_lfg::{ + handle_lfg_deactivate, handle_lfg_geo_aggregation, handle_lfg_get, handle_lfg_post, + handle_lfg_tags_autocomplete, + }, handle_host_meta::handle_host_meta, handle_import::{handle_import, handle_import_submit}, handle_index::handle_index, @@ -155,7 +159,13 @@ pub fn build_router(web_context: WebContext) -> Router { post(handle_xrpc_link_attestation), ) // API endpoints - .route("/api/geo-aggregation", get(handle_geo_aggregation)); + .route("/api/geo-aggregation", get(handle_geo_aggregation)) + .route("/api/lfg/tags", get(handle_lfg_tags_autocomplete)) + .route("/api/lfg/geo-aggregation", get(handle_lfg_geo_aggregation)) + // LFG routes + .route("/lfg", get(handle_lfg_get)) + .route("/lfg", post(handle_lfg_post)) + .route("/lfg/deactivate", post(handle_lfg_deactivate)); // Add OAuth metadata route only for AT Protocol backend if matches!( diff --git a/src/lib.rs b/src/lib.rs index 626cf5d..b3df9c3 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -28,6 +28,7 @@ pub mod stats; pub mod storage; pub mod tap_processor; pub mod task_identity_refresh; +pub mod task_lfg_cleanup; pub mod task_oauth_requests_cleanup; pub mod task_search_indexer; pub mod task_search_indexer_errors; diff --git a/src/search_index.rs b/src/search_index.rs index c6c7dc6..c683730 100644 --- a/src/search_index.rs +++ b/src/search_index.rs @@ -361,7 +361,7 @@ impl SearchIndexManager { } /// Index a single event - async fn index_event( + pub async fn index_event( &self, pool: &StoragePool, identity_resolver: Arc, @@ -931,6 +931,477 @@ impl SearchIndexManager { Ok(buckets) } + + /// Search for upcoming/ongoing events near a location. + /// + /// Returns events within the specified radius that are either: + /// - Starting in the future + /// - Currently ongoing (started but not ended) + /// + /// # Arguments + /// * `lat` - Center latitude + /// * `lon` - Center longitude + /// * `distance_miles` - Search radius in miles + /// * `limit` - Maximum number of results + pub async fn search_nearby_upcoming_events( + &self, + lat: f64, + lon: f64, + distance_miles: f64, + limit: u32, + ) -> Result> { + if !self.index_exists().await? { + return Ok(vec![]); + } + + let search_body = json!({ + "query": { + "bool": { + "must": [ + { + "geo_distance": { + "distance": format!("{}mi", distance_miles), + "locations_geo": { "lat": lat, "lon": lon } + } + } + ], + "should": [ + // Events starting in the future + { "range": { "start_time": { "gte": "now" } } }, + // Events currently ongoing + { + "bool": { + "must": [ + { "range": { "start_time": { "lte": "now" } } }, + { "range": { "end_time": { "gte": "now" } } } + ] + } + } + ], + "minimum_should_match": 1 + } + }, + "sort": [ + { "_geo_distance": { "locations_geo": { "lat": lat, "lon": lon }, "order": "asc" } }, + { "start_time": { "order": "asc" } } + ], + "size": limit + }); + + let response = self + .client + .search(SearchParts::Index(&[INDEX_NAME])) + .body(search_body) + .send() + .await?; + + if !response.status_code().is_success() { + return Err(anyhow::anyhow!("Nearby events search failed")); + } + + let body = response.json::().await?; + + let events: Vec = body["hits"]["hits"] + .as_array() + .map(|hits| { + hits.iter() + .filter_map(|hit| { + let source = &hit["_source"]; + serde_json::from_value(source.clone()).ok() + }) + .collect() + }) + .unwrap_or_default(); + + Ok(events) + } + + /// Get LFG profile geo aggregation for heatmap display. + /// + /// Returns H3 cell indices with profile counts at the specified precision. + pub async fn get_lfg_profile_geo_aggregation( + &self, + precision: u8, + center: Option, + ) -> Result> { + let agg_body = json!({ + "geohex_grid": { + "field": "location", + "precision": precision + } + }); + + // Build query filters - only active, non-expired profiles + let mut filters = vec![ + json!({ "term": { "active": true } }), + json!({ "range": { "ends_at": { "gte": "now" } } }), + ]; + + // Add geo_distance filter if center is provided + if let Some(c) = center { + filters.push(json!({ + "geo_distance": { + "distance": format!("{}mi", c.distance_miles), + "location": { "lat": c.lat, "lon": c.lon } + } + })); + } + + let search_body = json!({ + "size": 0, + "query": { + "bool": { "filter": filters } + }, + "aggs": { + "hex_grid": agg_body + } + }); + + let response = self + .client + .search(SearchParts::Index(&[Self::LFG_INDEX_NAME])) + .body(search_body) + .send() + .await?; + + if !response.status_code().is_success() { + return Err(anyhow::anyhow!("LFG geohex aggregation failed")); + } + + let body = response.json::().await?; + + let buckets = body["aggregations"]["hex_grid"]["buckets"] + .as_array() + .map(|arr| { + arr.iter() + .filter_map(|bucket| { + Some(GeoHexBucket { + key: bucket["key"].as_str()?.to_string(), + doc_count: bucket["doc_count"].as_u64()?, + }) + }) + .collect() + }) + .unwrap_or_default(); + + Ok(buckets) + } + + // ==================== LFG Profile Index Methods ==================== + + /// Index name for LFG profiles + const LFG_INDEX_NAME: &'static str = "smokesignal-lfg-profile"; + + /// Check if the LFG profile index exists + pub async fn lfg_index_exists(&self) -> Result { + let response = self + .client + .indices() + .exists(IndicesExistsParts::Index(&[Self::LFG_INDEX_NAME])) + .send() + .await?; + + Ok(response.status_code().is_success()) + } + + /// Create the LFG profile index with proper mappings + pub async fn create_lfg_profile_index(&self) -> Result<()> { + let exists = self.lfg_index_exists().await?; + + if exists { + tracing::debug!("Index {} already exists", Self::LFG_INDEX_NAME); + return Ok(()); + } + + let index_body = json!({ + "mappings": { + "properties": { + "aturi": { "type": "keyword" }, + "did": { "type": "keyword" }, + "location": { "type": "geo_point" }, + "tags": { "type": "keyword" }, + "starts_at": { "type": "date" }, + "ends_at": { "type": "date" }, + "active": { "type": "boolean" }, + "created_at": { "type": "date" } + } + } + }); + + let response = self + .client + .indices() + .create(IndicesCreateParts::Index(Self::LFG_INDEX_NAME)) + .body(index_body) + .send() + .await?; + + if !response.status_code().is_success() { + let error_body = response.text().await?; + return Err(anyhow::anyhow!("Failed to create LFG index: {}", error_body)); + } + + tracing::info!("Created OpenSearch index {}", Self::LFG_INDEX_NAME); + Ok(()) + } + + /// Index an LFG profile + pub async fn index_lfg_profile( + &self, + aturi: &str, + did: &str, + lat: f64, + lon: f64, + tags: &[String], + starts_at: &chrono::DateTime, + ends_at: &chrono::DateTime, + created_at: &chrono::DateTime, + active: bool, + ) -> Result<()> { + // Ensure index exists + self.create_lfg_profile_index().await?; + + let doc = json!({ + "aturi": aturi, + "did": did, + "location": { "lat": lat, "lon": lon }, + "tags": tags, + "starts_at": starts_at.to_rfc3339(), + "ends_at": ends_at.to_rfc3339(), + "created_at": created_at.to_rfc3339(), + "active": active + }); + + let response = self + .client + .index(IndexParts::IndexId(Self::LFG_INDEX_NAME, aturi)) + .body(doc) + .send() + .await?; + + if !response.status_code().is_success() { + let error_body = response.text().await?; + tracing::error!("Failed to index LFG profile {}: {}", aturi, error_body); + return Err(anyhow::anyhow!("Failed to index LFG profile")); + } + + Ok(()) + } + + /// Delete an LFG profile from the index + pub async fn delete_lfg_profile(&self, aturi: &str) -> Result<()> { + let response = self + .client + .delete(DeleteParts::IndexId(Self::LFG_INDEX_NAME, aturi)) + .send() + .await?; + + if !response.status_code().is_success() && response.status_code() != 404 { + return Err(anyhow::anyhow!("Failed to delete LFG profile")); + } + + Ok(()) + } + + /// Search for nearby LFG profiles + /// + /// Returns LFG profiles within the specified radius that have overlapping tags. + pub async fn search_nearby_lfg_profiles( + &self, + lat: f64, + lon: f64, + distance_miles: f64, + tags: &[String], + exclude_did: Option<&str>, + limit: u32, + ) -> Result> { + let must_clauses: Vec = vec![ + json!({ + "geo_distance": { + "distance": format!("{}mi", distance_miles), + "location": { "lat": lat, "lon": lon } + } + }), + json!({ "term": { "active": true } }), + json!({ "range": { "ends_at": { "gte": "now" } } }), + ]; + + let mut should_clauses = Vec::new(); + if !tags.is_empty() { + should_clauses.push(json!({ "terms": { "tags": tags } })); + } + + let mut must_not = Vec::new(); + if let Some(did) = exclude_did { + must_not.push(json!({ "term": { "did": did } })); + } + + let mut query = json!({ + "bool": { + "must": must_clauses + } + }); + + if !should_clauses.is_empty() { + query["bool"]["should"] = json!(should_clauses); + query["bool"]["minimum_should_match"] = json!(1); + } + + if !must_not.is_empty() { + query["bool"]["must_not"] = json!(must_not); + } + + let search_body = json!({ + "query": query, + "sort": [ + { "_score": { "order": "desc" } }, + { "_geo_distance": { "location": { "lat": lat, "lon": lon }, "order": "asc" } } + ], + "size": limit + }); + + let response = self + .client + .search(SearchParts::Index(&[Self::LFG_INDEX_NAME])) + .body(search_body) + .send() + .await?; + + if !response.status_code().is_success() { + return Err(anyhow::anyhow!("LFG profile search failed")); + } + + let body = response.json::().await?; + + let profiles: Vec = body["hits"]["hits"] + .as_array() + .map(|hits| { + hits.iter() + .filter_map(|hit| { + let source = &hit["_source"]; + serde_json::from_value(source.clone()).ok() + }) + .collect() + }) + .unwrap_or_default(); + + Ok(profiles) + } + + /// Get popular tags from active LFG profiles. + /// + /// Uses a terms aggregation to find the most commonly used tags across + /// all active (not expired) LFG profiles. + /// + /// Returns a vector of (tag, count) pairs sorted by count descending. + pub async fn get_popular_lfg_tags(&self, limit: u32) -> Result> { + let query = json!({ + "size": 0, + "query": { + "bool": { + "filter": [ + { "term": { "active": true } }, + { "range": { "ends_at": { "gt": "now" } } } + ] + } + }, + "aggs": { + "popular_tags": { + "terms": { + "field": "tags", + "size": limit + } + } + } + }); + + let response = self + .client + .search(opensearch::SearchParts::Index(&[Self::LFG_INDEX_NAME])) + .body(query) + .send() + .await?; + + if !response.status_code().is_success() { + let error_body = response.text().await?; + return Err(anyhow::anyhow!( + "Failed to get popular LFG tags: {}", + error_body + )); + } + + let body = response.json::().await?; + + let tags: Vec<(String, i64)> = body["aggregations"]["popular_tags"]["buckets"] + .as_array() + .map(|buckets| { + buckets + .iter() + .filter_map(|bucket| { + let key = bucket["key"].as_str()?.to_string(); + let count = bucket["doc_count"].as_i64()?; + Some((key, count)) + }) + .collect() + }) + .unwrap_or_default(); + + Ok(tags) + } + + /// Deactivate expired LFG profiles in the index + /// + /// This updates the `active` field to false for profiles where `ends_at` < now. + pub async fn deactivate_expired_lfg_profiles(&self) -> Result { + let update_body = json!({ + "script": { + "source": "ctx._source.active = false", + "lang": "painless" + }, + "query": { + "bool": { + "must": [ + { "term": { "active": true } }, + { "range": { "ends_at": { "lt": "now" } } } + ] + } + } + }); + + let response = self + .client + .update_by_query(opensearch::UpdateByQueryParts::Index(&[Self::LFG_INDEX_NAME])) + .body(update_body) + .send() + .await?; + + if !response.status_code().is_success() { + let error_body = response.text().await?; + return Err(anyhow::anyhow!("Failed to deactivate expired LFG profiles: {}", error_body)); + } + + let body = response.json::().await?; + let updated = body["updated"].as_u64().unwrap_or(0); + + if updated > 0 { + tracing::info!("Deactivated {} expired LFG profiles", updated); + } + + Ok(updated) + } +} + +/// Indexed LFG profile from OpenSearch +#[derive(Debug, Serialize, Deserialize)] +pub struct IndexedLfgProfile { + pub aturi: String, + pub did: String, + pub location: GeoPoint, + pub tags: Vec, + pub starts_at: String, + pub ends_at: String, + pub active: bool, + pub created_at: String, } #[cfg(test)] diff --git a/src/stats.rs b/src/stats.rs index 5630fd5..d24c112 100644 --- a/src/stats.rs +++ b/src/stats.rs @@ -28,6 +28,8 @@ pub(crate) enum StatsError { pub(crate) struct NetworkStats { pub event_count: i64, pub rsvp_count: i64, + pub lfg_identities_count: i64, + pub lfg_locations_count: i64, } impl NetworkStats { @@ -82,13 +84,19 @@ impl InMemoryCache { static IN_MEMORY_CACHE: once_cell::sync::Lazy>> = once_cell::sync::Lazy::new(|| Arc::new(RwLock::new(InMemoryCache::new()))); -/// Query database for event and RSVP counts +/// Query database for event, RSVP, and LFG counts async fn query_stats(pool: &StoragePool) -> Result { let row = sqlx::query( r#" SELECT (SELECT COUNT(*) FROM events) as event_count, - (SELECT COUNT(*) FROM rsvps) as rsvp_count + (SELECT COUNT(*) FROM rsvps) as rsvp_count, + (SELECT COUNT(DISTINCT did) FROM atproto_records + WHERE collection = 'events.smokesignal.lfg' + AND (record->>'active')::boolean = true) as lfg_identities_count, + (SELECT COUNT(DISTINCT record->'location'->>'value') FROM atproto_records + WHERE collection = 'events.smokesignal.lfg' + AND (record->>'active')::boolean = true) as lfg_locations_count "#, ) .fetch_one(pool) @@ -102,6 +110,12 @@ async fn query_stats(pool: &StoragePool) -> Result { rsvp_count: row .try_get("rsvp_count") .map_err(|e| StatsError::DatabaseError(e.to_string()))?, + lfg_identities_count: row + .try_get("lfg_identities_count") + .map_err(|e| StatsError::DatabaseError(e.to_string()))?, + lfg_locations_count: row + .try_get("lfg_locations_count") + .map_err(|e| StatsError::DatabaseError(e.to_string()))?, }) } diff --git a/src/storage/atproto_record.rs b/src/storage/atproto_record.rs index 5e5a71b..49c51d3 100644 --- a/src/storage/atproto_record.rs +++ b/src/storage/atproto_record.rs @@ -183,10 +183,11 @@ pub async fn atproto_record_get_location_suggestions( .await .map_err(StorageError::UnableToExecuteQuery)?; - // Collect all suggestions with timestamps for sorting - let mut suggestions_with_time: Vec<(DateTime, LocationSuggestion)> = Vec::new(); + // Collect all suggestions with (priority, timestamp) for sorting + // Priority: 0 = beaconbits/dropanchor (higher priority), 1 = smokesignal events + let mut suggestions_with_priority: Vec<(u8, DateTime, LocationSuggestion)> = Vec::new(); - // Process atproto records (beaconbits/dropanchor) + // Process atproto records (beaconbits/dropanchor) - priority 0 for (aturi, indexed_at, record) in atproto_rows { let r = &record.0; // Try addressDetails (beaconbits) first, then address (dropanchor) @@ -229,25 +230,30 @@ pub async fn atproto_record_get_location_suggestions( .and_then(|v| v.as_str()) .map(String::from), }; - suggestions_with_time.push((indexed_at, suggestion)); + suggestions_with_priority.push((0, indexed_at, suggestion)); } - // Process event locations + // Process event locations - priority 1 (lower than beaconbits/dropanchor) for (aturi, updated_at, record) in event_rows { let timestamp = updated_at.unwrap_or_else(Utc::now); let locations = extract_locations_from_event(&aturi, &record.0); for suggestion in locations { - suggestions_with_time.push((timestamp, suggestion)); + suggestions_with_priority.push((1, timestamp, suggestion)); } } - // Sort by timestamp descending (most recent first) - suggestions_with_time.sort_by(|a, b| b.0.cmp(&a.0)); + // Sort by priority ascending (0 first), then by timestamp descending + suggestions_with_priority.sort_by(|a, b| { + match a.0.cmp(&b.0) { + std::cmp::Ordering::Equal => b.1.cmp(&a.1), // Same priority: newer first + other => other, // Different priority: lower first + } + }); // Collect into OrderSet (deduplicates based on location fields) - let suggestions: OrderSet = suggestions_with_time + let suggestions: OrderSet = suggestions_with_priority .into_iter() - .map(|(_, s)| s) + .map(|(_, _, s)| s) .collect(); Ok(suggestions) diff --git a/src/storage/lfg.rs b/src/storage/lfg.rs new file mode 100644 index 0000000..015ca04 --- /dev/null +++ b/src/storage/lfg.rs @@ -0,0 +1,94 @@ +//! Storage module for LFG (Looking For Group) records. +//! +//! This module provides query helpers for LFG records stored in the `atproto_records` table. +//! For writes, use `atproto_record_upsert()` and `atproto_record_delete()` from the +//! `atproto_record` module. +//! +//! Records are stored as `AtprotoRecord` and can be deserialized to `Lfg` using: +//! ```ignore +//! let lfg: Lfg = serde_json::from_value(record.record.0.clone())?; +//! ``` + +use super::atproto_record::AtprotoRecord; +use super::errors::StorageError; +use super::StoragePool; +use crate::atproto::lexicon::lfg::NSID; + +/// Get the active LFG record for a DID from atproto_records. +/// +/// Returns the most recent active LFG record for the given DID, or None if +/// no active record exists. +pub async fn lfg_get_active_by_did( + pool: &StoragePool, + did: &str, +) -> Result, StorageError> { + let record = sqlx::query_as::<_, AtprotoRecord>( + r#" + SELECT aturi, did, cid, collection, indexed_at, record + FROM atproto_records + WHERE did = $1 + AND collection = $2 + AND (record->>'active')::boolean = true + AND (record->>'endsAt')::timestamptz > NOW() + ORDER BY indexed_at DESC + LIMIT 1 + "#, + ) + .bind(did) + .bind(NSID) + .fetch_optional(pool) + .await + .map_err(StorageError::UnableToExecuteQuery)?; + + Ok(record) +} + +/// Get an LFG record by AT-URI from atproto_records. +pub async fn lfg_get_by_aturi( + pool: &StoragePool, + aturi: &str, +) -> Result, StorageError> { + let record = sqlx::query_as::<_, AtprotoRecord>( + r#" + SELECT aturi, did, cid, collection, indexed_at, record + FROM atproto_records + WHERE aturi = $1 + AND collection = $2 + "#, + ) + .bind(aturi) + .bind(NSID) + .fetch_optional(pool) + .await + .map_err(StorageError::UnableToExecuteQuery)?; + + Ok(record) +} + +/// Get all LFG records for a DID (including inactive/expired). +/// +/// Used for tag history lookups. +pub async fn lfg_get_all_by_did( + pool: &StoragePool, + did: &str, + limit: i64, +) -> Result, StorageError> { + let records = sqlx::query_as::<_, AtprotoRecord>( + r#" + SELECT aturi, did, cid, collection, indexed_at, record + FROM atproto_records + WHERE did = $1 + AND collection = $2 + ORDER BY indexed_at DESC + LIMIT $3 + "#, + ) + .bind(did) + .bind(NSID) + .bind(limit) + .fetch_all(pool) + .await + .map_err(StorageError::UnableToExecuteQuery)?; + + Ok(records) +} diff --git a/src/storage/mod.rs b/src/storage/mod.rs index 8f43ac9..b3fe25e 100644 --- a/src/storage/mod.rs +++ b/src/storage/mod.rs @@ -8,6 +8,7 @@ pub mod distributed_lock; pub mod errors; pub mod event; pub mod identity_profile; +pub mod lfg; pub mod notification; pub mod oauth; pub mod private_event_content; diff --git a/src/tap_processor.rs b/src/tap_processor.rs index 894fc21..e22d324 100644 --- a/src/tap_processor.rs +++ b/src/tap_processor.rs @@ -207,6 +207,7 @@ impl TapProcessor { | "community.lexicon.calendar.event" | "events.smokesignal.profile" | "events.smokesignal.calendar.acceptance" + | "events.smokesignal.lfg" | "app.beaconbits.bookmark.item" | "app.beaconbits.beacon" | "app.dropanchor.checkin" diff --git a/src/task_lfg_cleanup.rs b/src/task_lfg_cleanup.rs new file mode 100644 index 0000000..8fa0f00 --- /dev/null +++ b/src/task_lfg_cleanup.rs @@ -0,0 +1,163 @@ +//! LFG (Looking For Group) cleanup background task. +//! +//! This task runs periodically to deactivate expired LFG records in both +//! the database and OpenSearch index. + +use anyhow::Result; +use chrono::{Duration, Utc}; +use tokio::time::{Instant, sleep}; +use tokio_util::sync::CancellationToken; + +use crate::search_index::SearchIndexManager; +use crate::storage::StoragePool; +use crate::atproto::lexicon::lfg::NSID; + +/// Configuration for the LFG cleanup task. +pub struct LfgCleanupTaskConfig { + /// How often to run the cleanup (default: 1 hour) + pub sleep_interval: Duration, +} + +impl Default for LfgCleanupTaskConfig { + fn default() -> Self { + Self { + sleep_interval: Duration::hours(1), + } + } +} + +/// Background task that deactivates expired LFG records. +pub struct LfgCleanupTask { + pub config: LfgCleanupTaskConfig, + pub storage_pool: StoragePool, + pub search_index: Option, + pub cancellation_token: CancellationToken, +} + +impl LfgCleanupTask { + /// Creates a new LFG cleanup task. + #[must_use] + pub fn new( + config: LfgCleanupTaskConfig, + storage_pool: StoragePool, + search_index: Option, + cancellation_token: CancellationToken, + ) -> Self { + Self { + config, + storage_pool, + search_index, + cancellation_token, + } + } + + /// Runs the LFG cleanup task as a long-running process. + /// + /// This task: + /// 1. Deactivates expired LFG records in the database + /// 2. Updates the OpenSearch index to reflect expired records + /// + /// # Errors + /// Returns an error if the sleep interval cannot be converted, or if there's + /// a problem cleaning up expired records. + pub async fn run(&self) -> Result<()> { + tracing::info!("LfgCleanupTask started"); + + let interval = self.config.sleep_interval.to_std()?; + + let sleeper = sleep(interval); + tokio::pin!(sleeper); + + loop { + tokio::select! { + () = self.cancellation_token.cancelled() => { + break; + }, + () = &mut sleeper => { + if let Err(err) = self.cleanup_expired_lfg_records().await { + tracing::error!("LfgCleanupTask failed: {}", err); + } + sleeper.as_mut().reset(Instant::now() + interval); + } + } + } + + tracing::info!("LfgCleanupTask stopped"); + + Ok(()) + } + + /// Cleanup expired LFG records. + async fn cleanup_expired_lfg_records(&self) -> Result<()> { + let now = Utc::now(); + + tracing::debug!("Starting cleanup of expired LFG records"); + + // Step 1: Update expired records in the database + let db_result = self.deactivate_expired_in_database(&now).await?; + + // Step 2: Update expired records in OpenSearch + let os_result = self.deactivate_expired_in_opensearch().await?; + + if db_result > 0 || os_result > 0 { + tracing::info!( + database_updated = db_result, + opensearch_updated = os_result, + "Cleaned up expired LFG records" + ); + } else { + tracing::debug!("No expired LFG records to clean up"); + } + + Ok(()) + } + + /// Deactivate expired LFG records in the database. + async fn deactivate_expired_in_database( + &self, + now: &chrono::DateTime, + ) -> Result { + // Query for active LFG records that have expired + let result = sqlx::query( + r#" + UPDATE atproto_records + SET record = jsonb_set(record, '{active}', 'false') + WHERE collection = $1 + AND (record->>'active')::boolean = true + AND (record->>'endsAt')::timestamptz < $2 + "#, + ) + .bind(NSID) + .bind(now) + .execute(&self.storage_pool) + .await?; + + Ok(result.rows_affected()) + } + + /// Deactivate expired LFG profiles in OpenSearch. + async fn deactivate_expired_in_opensearch(&self) -> Result { + let Some(ref search_index) = self.search_index else { + return Ok(0); + }; + + match search_index.deactivate_expired_lfg_profiles().await { + Ok(count) => Ok(count), + Err(err) => { + tracing::warn!("Failed to deactivate expired LFG profiles in OpenSearch: {}", err); + Ok(0) + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_default_config() { + let config = LfgCleanupTaskConfig::default(); + assert_eq!(config.sleep_interval, Duration::hours(1)); + } +} diff --git a/src/task_search_indexer.rs b/src/task_search_indexer.rs index edfd8cc..09f962b 100644 --- a/src/task_search_indexer.rs +++ b/src/task_search_indexer.rs @@ -1,19 +1,20 @@ use anyhow::Result; -use atproto_attestation::create_dagbor_cid; use atproto_identity::{model::Document, resolve::IdentityResolver, traits::DidDocumentStorage}; -use atproto_record::lexicon::app::bsky::richtext::facet::{Facet, FacetFeature}; use atproto_record::lexicon::community::lexicon::calendar::event::NSID as LexiconCommunityEventNSID; use opensearch::{ DeleteParts, IndexParts, OpenSearch, http::transport::Transport, indices::{IndicesCreateParts, IndicesExistsParts}, }; -use serde::Deserialize; use serde_json::{Value, json}; use std::sync::Arc; +use crate::atproto::lexicon::lfg::{Lfg, NSID as LFG_NSID}; use crate::atproto::lexicon::profile::{Profile, NSID as PROFILE_NSID}; use crate::atproto::utils::get_profile_hashtags; +use crate::search_index::SearchIndexManager; +use crate::storage::event::event_get; +use crate::storage::StoragePool; use crate::task_search_indexer_errors::SearchIndexerError; /// Build an AT URI with pre-allocated capacity to avoid format! overhead. @@ -30,68 +31,16 @@ fn build_aturi(did: &str, nsid: &str, rkey: &str) -> String { uri } -/// Generate a DAG-CBOR CID from a JSON value. -/// -/// Creates a CIDv1 with DAG-CBOR codec (0x71) and SHA-256 hash (0x12), -/// following the AT Protocol specification for content addressing. -fn generate_location_cid(value: &Value) -> Result { - let cid = create_dagbor_cid(value).map_err(|e| SearchIndexerError::CidGenerationFailed { - error: e.to_string(), - })?; - Ok(cid.to_string()) -} - -/// A lightweight event struct for search indexing. -/// -/// Uses serde_json::Value for locations to avoid deserialization errors -/// when event data contains location types not supported by the LocationOrRef enum. -#[derive(Deserialize)] -struct IndexableEvent { - name: String, - description: String, - #[serde(rename = "createdAt")] - created_at: chrono::DateTime, - #[serde(rename = "startsAt")] - starts_at: Option>, - #[serde(rename = "endsAt")] - ends_at: Option>, - #[serde(rename = "descriptionFacets")] - facets: Option>, - /// Locations stored as raw JSON values for CID generation. - #[serde(default)] - locations: Vec, -} - -impl IndexableEvent { - /// Extract hashtags from the event's facets - fn get_hashtags(&self) -> Vec { - self.facets - .as_ref() - .map(|facets| { - facets - .iter() - .flat_map(|facet| { - facet.features.iter().filter_map(|feature| { - if let FacetFeature::Tag(tag) = feature { - Some(tag.tag.clone()) - } else { - None - } - }) - }) - .collect() - }) - .unwrap_or_default() - } -} - const EVENTS_INDEX_NAME: &str = "smokesignal-events"; const PROFILES_INDEX_NAME: &str = "smokesignal-profiles"; +const LFG_INDEX_NAME: &str = "smokesignal-lfg-profile"; pub struct SearchIndexer { client: Arc, + pool: StoragePool, identity_resolver: Arc, document_storage: Arc, + event_index_manager: SearchIndexManager, } impl SearchIndexer { @@ -100,20 +49,25 @@ impl SearchIndexer { /// # Arguments /// /// * `endpoint` - OpenSearch endpoint URL + /// * `pool` - Database connection pool for fetching events /// * `identity_resolver` - Resolver for DID identities /// * `document_storage` - Storage for DID documents pub async fn new( endpoint: &str, + pool: StoragePool, identity_resolver: Arc, document_storage: Arc, ) -> Result { let transport = Transport::single_node(endpoint)?; let client = Arc::new(OpenSearch::new(transport)); + let event_index_manager = SearchIndexManager::new(endpoint)?; let indexer = Self { client, + pool, identity_resolver, document_storage, + event_index_manager, }; indexer.ensure_index().await?; @@ -126,6 +80,8 @@ impl SearchIndexer { self.ensure_events_index().await?; // Ensure profiles index self.ensure_profiles_index().await?; + // Ensure LFG profiles index + self.ensure_lfg_profiles_index().await?; Ok(()) } @@ -151,6 +107,7 @@ impl SearchIndexer { "description": { "type": "text" }, "tags": { "type": "keyword" }, "location_cids": { "type": "keyword" }, + "locations_geo": { "type": "geo_point" }, "start_time": { "type": "date" }, "end_time": { "type": "date" }, "created_at": { "type": "date" }, @@ -222,6 +179,52 @@ impl SearchIndexer { Ok(()) } + async fn ensure_lfg_profiles_index(&self) -> Result<()> { + let exists_response = self + .client + .indices() + .exists(IndicesExistsParts::Index(&[LFG_INDEX_NAME])) + .send() + .await?; + + if exists_response.status_code().is_success() { + tracing::info!("OpenSearch index {} already exists", LFG_INDEX_NAME); + return Ok(()); + } + + let index_body = json!({ + "mappings": { + "properties": { + "aturi": { "type": "keyword" }, + "did": { "type": "keyword" }, + "location": { "type": "geo_point" }, + "tags": { "type": "keyword" }, + "starts_at": { "type": "date" }, + "ends_at": { "type": "date" }, + "active": { "type": "boolean" }, + "created_at": { "type": "date" } + } + }, + }); + + let response = self + .client + .indices() + .create(IndicesCreateParts::Index(LFG_INDEX_NAME)) + .body(index_body) + .send() + .await?; + + if response.status_code().is_success() { + tracing::info!("Created OpenSearch index {}", LFG_INDEX_NAME); + } else { + let error_body = response.text().await?; + return Err(SearchIndexerError::IndexCreationFailed { error_body }.into()); + } + + Ok(()) + } + /// Index a commit event (create or update). /// /// Dispatches to the appropriate indexer based on collection type. @@ -236,6 +239,7 @@ impl SearchIndexer { match collection { "community.lexicon.calendar.event" => self.index_event(did, rkey, record).await, c if c == PROFILE_NSID => self.index_profile(did, rkey, record).await, + c if c == LFG_NSID => self.index_lfg_profile(did, rkey, record).await, _ => Ok(()), } } @@ -247,66 +251,31 @@ impl SearchIndexer { match collection { "community.lexicon.calendar.event" => self.delete_event(did, rkey).await, c if c == PROFILE_NSID => self.delete_profile(did, rkey).await, + c if c == LFG_NSID => self.delete_lfg_profile(did, rkey).await, _ => Ok(()), } } - async fn index_event(&self, did: &str, rkey: &str, record: Value) -> Result<()> { - let event: IndexableEvent = serde_json::from_value(record)?; - - let document = self.ensure_identity_stored(did).await?; - let handle = document.handles().unwrap_or("invalid.handle"); - + async fn index_event(&self, did: &str, rkey: &str, _record: Value) -> Result<()> { let aturi = build_aturi(did, LexiconCommunityEventNSID, rkey); - // Extract fields from the IndexableEvent struct - let name = &event.name; - let description = &event.description; - let created_at = &event.created_at; - let starts_at = &event.starts_at; - let ends_at = &event.ends_at; - - // Extract hashtags from facets - let tags = event.get_hashtags(); - - // Generate CIDs for each location - let location_cids: Vec = event - .locations - .iter() - .filter_map(|loc| generate_location_cid(loc).ok()) - .collect(); - - let mut doc = json!({ - "did": did, - "handle": handle, - "name": name, - "description": description, - "tags": tags, - "location_cids": location_cids, - "created_at": json!(created_at), - "updated_at": json!(chrono::Utc::now()) - }); - - // Add optional time fields - if let Some(start) = starts_at { - doc["start_time"] = json!(start); - } - if let Some(end) = ends_at { - doc["end_time"] = json!(end); - } - - let response = self - .client - .index(IndexParts::IndexId(EVENTS_INDEX_NAME, &aturi)) - .body(doc) - .send() - .await?; - - if response.status_code().is_success() { - tracing::debug!("Indexed event {} for DID {}", rkey, did); - } else { - let error_body = response.text().await?; - tracing::error!("Failed to index event: {}", error_body); + // Fetch the event from the database and delegate to SearchIndexManager + // This ensures we use the same indexing logic as the web handlers + match event_get(&self.pool, &aturi).await { + Ok(event) => { + self.event_index_manager + .index_event(&self.pool, self.identity_resolver.clone(), &event) + .await?; + tracing::debug!("Indexed event {} for DID {}", rkey, did); + } + Err(err) => { + // Event might not be in the database yet if content fetcher hasn't processed it + tracing::warn!( + "Could not fetch event {} for indexing: {}. It may be indexed on next update.", + aturi, + err + ); + } } Ok(()) @@ -315,19 +284,10 @@ impl SearchIndexer { async fn delete_event(&self, did: &str, rkey: &str) -> Result<()> { let aturi = build_aturi(did, LexiconCommunityEventNSID, rkey); - let response = self - .client - .delete(DeleteParts::IndexId(EVENTS_INDEX_NAME, &aturi)) - .send() - .await?; - - if response.status_code().is_success() || response.status_code() == 404 { - tracing::debug!("Deleted event {} for DID {} from search index", rkey, did); - } else { - let error_body = response.text().await?; - tracing::error!("Failed to delete event from index: {}", error_body); - } + // Delegate to SearchIndexManager for consistent deletion logic + self.event_index_manager.delete_indexed_event(&aturi).await?; + tracing::debug!("Deleted event {} for DID {} from search index", rkey, did); Ok(()) } @@ -392,6 +352,42 @@ impl SearchIndexer { Ok(()) } + async fn index_lfg_profile(&self, did: &str, rkey: &str, record: Value) -> Result<()> { + let lfg: Lfg = serde_json::from_value(record)?; + let aturi = build_aturi(did, LFG_NSID, rkey); + + // Extract coordinates from location + let (lat, lon) = lfg.get_coordinates().unwrap_or((0.0, 0.0)); + + // Delegate to SearchIndexManager for consistent indexing logic + self.event_index_manager + .index_lfg_profile( + &aturi, + did, + lat, + lon, + &lfg.tags, + &lfg.starts_at, + &lfg.ends_at, + &lfg.created_at, + lfg.active, + ) + .await?; + + tracing::debug!("Indexed LFG profile {} for DID {}", rkey, did); + Ok(()) + } + + async fn delete_lfg_profile(&self, did: &str, rkey: &str) -> Result<()> { + let aturi = build_aturi(did, LFG_NSID, rkey); + + // Delegate to SearchIndexManager for consistent deletion logic + self.event_index_manager.delete_lfg_profile(&aturi).await?; + + tracing::debug!("Deleted LFG profile {} for DID {} from search index", rkey, did); + Ok(()) + } + async fn ensure_identity_stored(&self, did: &str) -> Result { // Check if we already have this identity if let Some(document) = self.document_storage.get_document_by_did(did).await? { diff --git a/src/task_search_indexer_errors.rs b/src/task_search_indexer_errors.rs index fa2f6ba..0baca69 100644 --- a/src/task_search_indexer_errors.rs +++ b/src/task_search_indexer_errors.rs @@ -12,11 +12,4 @@ pub(crate) enum SearchIndexerError { /// and the operation fails with a server error response. #[error("error-smokesignal-search-indexer-1 Failed to create index: {error_body}")] IndexCreationFailed { error_body: String }, - - /// Error when CID generation fails. - /// - /// This error occurs when serializing location data to DAG-CBOR - /// or generating the multihash for a CID fails. - #[error("error-smokesignal-search-indexer-2 Failed to generate CID: {error}")] - CidGenerationFailed { error: String }, } diff --git a/templates/en-us/create_event.alpine.html b/templates/en-us/create_event.alpine.html index 4f7cb49..7f86ee2 100644 --- a/templates/en-us/create_event.alpine.html +++ b/templates/en-us/create_event.alpine.html @@ -768,7 +768,19 @@ function eventForm() { init() { // Load existing location data from server if present - {% if location_form.location_country %} + {% if event_locations %} + {% for loc in event_locations %} + this.formData.locations.push({ + country: {{ loc.country | tojson }}, + postal_code: {{ loc.postal_code | tojson }}, + region: {{ loc.region | tojson }}, + locality: {{ loc.locality | tojson }}, + street: {{ loc.street | tojson }}, + name: {{ loc.name | tojson }}, + }); + {% endfor %} + {% elif location_form.location_country %} + // Fallback to old single-location format this.formData.locations.push({ country: {{ location_form.location_country | tojson }}, postal_code: {% if location_form.location_postal_code %}{{ location_form.location_postal_code | tojson }}{% else %}null{% endif %}, @@ -779,6 +791,17 @@ function eventForm() { }); {% endif %} + // Load existing geo locations from server if present + {% if event_geo_locations %} + {% for geo in event_geo_locations %} + this.formData.geo_locations.push({ + latitude: {{ geo.latitude | tojson }}, + longitude: {{ geo.longitude | tojson }}, + name: {{ geo.name | tojson }}, + }); + {% endfor %} + {% endif %} + // Load existing links from server if present {% if event_links %} {% for link in event_links %} diff --git a/templates/en-us/event_list.incl.html b/templates/en-us/event_list.incl.html index c4e5f17..991306e 100644 --- a/templates/en-us/event_list.incl.html +++ b/templates/en-us/event_list.incl.html @@ -158,6 +158,16 @@

{% autoescape false %}{{ event.description_short }}{% endautoescape %}

+ {% if user_tags and event.description_tags %} +
+ {% for tag in event.description_tags %} + {% if tag | lower in user_tags %} + {{ tag }} + {% endif %} + {% endfor %} +
+ {% endif %} + diff --git a/templates/en-us/index.common.html b/templates/en-us/index.common.html index 0e98cc3..da0e8ef 100644 --- a/templates/en-us/index.common.html +++ b/templates/en-us/index.common.html @@ -48,7 +48,7 @@ - A network of {{ event_count }} events and {{ rsvp_count }} RSVPs + A network of {{ event_count }} events and {{ rsvp_count }} RSVPs with {{ lfg_identities_count }} people LFG accross {{ lfg_locations_count }} locations.

@@ -278,7 +278,7 @@ // Default center (world view) if geolocation fails const DEFAULT_CENTER = [39.8283, -98.5795]; // Center of USA - const DEFAULT_ZOOM = 9; + const DEFAULT_ZOOM = 10; // Color scale for heatmap (low to high) const colors = ['#ffffcc', '#ffeda0', '#fed976', '#feb24c', '#fd8d3c', '#fc4e2a', '#e31a1c', '#bd0026', '#800026']; diff --git a/templates/en-us/lfg_form.bare.html b/templates/en-us/lfg_form.bare.html new file mode 100644 index 0000000..a500302 --- /dev/null +++ b/templates/en-us/lfg_form.bare.html @@ -0,0 +1,4 @@ +{% extends "en-us/bare.html" %} +{% block content %} +{% include 'en-us/lfg_form.common.html' %} +{% endblock %} diff --git a/templates/en-us/lfg_form.common.html b/templates/en-us/lfg_form.common.html new file mode 100644 index 0000000..f5c74af --- /dev/null +++ b/templates/en-us/lfg_form.common.html @@ -0,0 +1,329 @@ +
+
+

+ + Looking For Group +

+

Find activity partners in your area

+ +
+
+ + +
+ +

Click on the map to select your general area. Click again on the highlighted area to unselect.

+ +
+ +

+ + Location selected +

+

+
+ + +
+ +

Add 1-10 tags describing what you're looking for

+ +
+ +
+ + +
+ +
+ +

+ + + {% if popular_tags %} +
+

Popular tags:

+
+ {% for tag in popular_tags %} + + {% endfor %} +
+
+ {% endif %} +
+ + +
+ +

How long should your LFG be active?

+ +
+
+ +
+
+ +

+
+ + +
+ +
+ + +
+
+ +
+
+ Cancel +
+
+
+
+
+
+ + diff --git a/templates/en-us/lfg_form.html b/templates/en-us/lfg_form.html new file mode 100644 index 0000000..8e8d15c --- /dev/null +++ b/templates/en-us/lfg_form.html @@ -0,0 +1,15 @@ +{% extends "en-us/base.html" %} +{% block title %}Looking For Group - Smoke Signal{% endblock %} +{% block head %} + + + + + + + + +{% endblock %} +{% block content %} +{% include 'en-us/lfg_form.common.html' %} +{% endblock %} diff --git a/templates/en-us/lfg_matches.bare.html b/templates/en-us/lfg_matches.bare.html new file mode 100644 index 0000000..adc86c3 --- /dev/null +++ b/templates/en-us/lfg_matches.bare.html @@ -0,0 +1,4 @@ +{% extends "en-us/bare.html" %} +{% block content %} +{% include 'en-us/lfg_matches.common.html' %} +{% endblock %} diff --git a/templates/en-us/lfg_matches.common.html b/templates/en-us/lfg_matches.common.html new file mode 100644 index 0000000..daa4b42 --- /dev/null +++ b/templates/en-us/lfg_matches.common.html @@ -0,0 +1,194 @@ +
+
+
+
+
+
+

+ + Active +

+
+
+
+
+
+ +
+
+
+
+ + {% if lfg_record %} +
+

+ {% for tag in lfg_record.tags %} + {{ tag }} + {% endfor %} + {% if h3_cell %} + + + {{ h3_cell }} + + {% endif %} +

+

+ Expires at +

+
+ {% endif %} +
+ + +
+

+ + Activity +

+
+
+ + +
+ +
+
+

+ + Events + {{ events | length }} +

+ + {% if events | length > 0 %} + {% set base = "" %} + {% include 'en-us/event_list.incl.html' %} + {% else %} +

+ No matching events found nearby. Check back later or adjust your tags. +

+ {% endif %} +
+
+ + +
+
+

+ + People + {{ matching_profiles | length }} +

+ + {% if matching_profiles | length > 0 %} +
+ {% for profile in matching_profiles %} +
+
+

+ + + {% if profile.display_name %} + {{ profile.display_name }} + {% elif profile.handle %} + @{{ profile.handle }} + {% else %} + {{ profile.did }} + {% endif %} + + {% if profile.display_name and profile.handle %} +
+ @{{ profile.handle }} + {% endif %} +
+ + {% for tag in profile.tags %} + {% if tag | lower in user_tags %} + {{ tag }} + {% else %} + {{ tag }} + {% endif %} + {% endfor %} + +

+
+
+ {% endfor %} +
+ {% else %} +

+ No other people looking in your area yet. Share your profile to connect! +

+ {% endif %} +
+
+
+
+
+ + diff --git a/templates/en-us/lfg_matches.html b/templates/en-us/lfg_matches.html new file mode 100644 index 0000000..4c35949 --- /dev/null +++ b/templates/en-us/lfg_matches.html @@ -0,0 +1,15 @@ +{% extends "en-us/base.html" %} +{% block title %}Looking For Group - Smoke Signal{% endblock %} +{% block head %} + + + + + + + + +{% endblock %} +{% block content %} +{% include 'en-us/lfg_matches.common.html' %} +{% endblock %} diff --git a/templates/en-us/nav.html b/templates/en-us/nav.html index cae5729..3fd073a 100644 --- a/templates/en-us/nav.html +++ b/templates/en-us/nav.html @@ -28,6 +28,12 @@ Search + + + + + LFG + Help