diff --git a/consumer/src/database_writer/workers.rs b/consumer/src/database_writer/workers.rs index fe8c1c9d..4595599c 100644 --- a/consumer/src/database_writer/workers.rs +++ b/consumer/src/database_writer/workers.rs @@ -520,70 +520,38 @@ async fn batch_update_actor_stats( for (batch_idx, chunk) in deltas_vec.chunks(BATCH_SIZE).enumerate() { let batch_start = std::time::Instant::now(); - // Build CASE expression components - let mut case_followers = String::new(); - let mut case_following = String::new(); - let mut case_posts = String::new(); - let mut case_lists = String::new(); - let mut case_feeds = String::new(); - let mut case_starterpacks = String::new(); - let mut id_list = String::new(); - - for (i, (actor_id, delta)) in chunk.iter().enumerate() { - if i > 0 { - case_followers.push(' '); - case_following.push(' '); - case_posts.push(' '); - case_lists.push(' '); - case_feeds.push(' '); - case_starterpacks.push(' '); - id_list.push_str(", "); - } + // Convert to ActorUpdate format + let mut count_deltas = std::collections::HashMap::new(); + let mut actor_ids = Vec::new(); - // CASE clauses for each actor - case_followers.push_str(&format!( - "WHEN {} THEN NULLIF(LEAST(2147483647, GREATEST(0, COALESCE(followers_count, 0) + {})), 0)", - actor_id, delta.followers_delta - )); - case_following.push_str(&format!( - "WHEN {} THEN NULLIF(LEAST(2147483647, GREATEST(0, COALESCE(following_count, 0) + {})), 0)", - actor_id, delta.following_delta - )); - case_posts.push_str(&format!( - "WHEN {} THEN NULLIF(LEAST(2147483647, GREATEST(0, COALESCE(posts_count, 0) + {})), 0)", - actor_id, delta.posts_delta - )); - case_lists.push_str(&format!( - "WHEN {} THEN NULLIF(LEAST(32767, GREATEST(0, COALESCE(lists_count, 0) + {}))::smallint, 0)", - actor_id, delta.lists_delta - )); - case_feeds.push_str(&format!( - "WHEN {} THEN NULLIF(LEAST(32767, GREATEST(0, COALESCE(feeds_count, 0) + {}))::smallint, 0)", - actor_id, delta.feeds_delta - )); - case_starterpacks.push_str(&format!( - "WHEN {} THEN NULLIF(LEAST(32767, GREATEST(0, COALESCE(starterpacks_count, 0) + {}))::smallint, 0)", - actor_id, delta.starterpacks_delta - )); - - id_list.push_str(&format!("{}", actor_id)); + for (actor_id, delta) in chunk { + actor_ids.push(*actor_id); + count_deltas.insert( + *actor_id, + crate::db::operations::ActorCountDeltas { + followers_count: Some(delta.followers_delta), + following_count: Some(delta.following_delta), + posts_count: Some(delta.posts_delta), + lists_count: Some(delta.lists_delta as i32), + feeds_count: Some(delta.feeds_delta as i32), + starterpacks_count: Some(delta.starterpacks_delta as i32), + }, + ); } - // Execute batched UPDATE with CASE expressions - let query = format!( - "UPDATE actors - SET - followers_count = CASE id {} END, - following_count = CASE id {} END, - posts_count = CASE id {} END, - lists_count = CASE id {} END, - feeds_count = CASE id {} END, - starterpacks_count = CASE id {} END - WHERE id IN ({})", - case_followers, case_following, case_posts, case_lists, case_feeds, case_starterpacks, id_list - ); + // Use consolidated ActorUpdate API + let result = crate::db::operations::ActorUpdate { + target: crate::db::operations::ActorUpdateTarget::Batch { ids: actor_ids }, + count_deltas: Some(count_deltas), + ..Default::default() + } + .execute(conn) + .await?; - let updated = conn.execute(&query, &[]).await?; + let updated = match result { + crate::db::operations::ActorUpdateResult::Count(n) => n, + _ => unreachable!("ActorUpdate with Count returning should return Count"), + }; total_updated += updated; tracing::debug!( diff --git a/consumer/src/db/actor.rs b/consumer/src/db/actor.rs index cf773090..d377238c 100644 --- a/consumer/src/db/actor.rs +++ b/consumer/src/db/actor.rs @@ -70,12 +70,23 @@ pub async fn actor_set_sync_status( sync_state: &ActorSyncState, time: DateTime, ) -> Result { - conn.execute( - "UPDATE actors SET sync_state=$2, last_indexed=$3 WHERE did=$1", - &[&did, &sync_state, &time], - ) + // Use consolidated ActorUpdate API for sync status update + use crate::db::operations::{ActorUpdate, ActorUpdateResult, ActorUpdateTarget}; + + let result = ActorUpdate { + target: ActorUpdateTarget::ByDid(did.to_string()), + sync_state: Some(*sync_state), + last_indexed: Some(time), + ..Default::default() + } + .execute(conn) .await - .wrap_err_with(|| format!("Failed to set sync status for actor {}", did)) + .wrap_err_with(|| format!("Failed to set sync status for actor {}", did))?; + + match result { + ActorUpdateResult::Count(n) => Ok(n), + _ => unreachable!("ActorUpdate with Count returning should return Count"), + } } pub async fn actor_set_repo_state( @@ -88,12 +99,23 @@ pub async fn actor_set_repo_state( let cid_digest = parakeet_db::cid_util::cid_to_digest(&cid_bytes) .expect("CID must be valid AT Protocol CID"); - conn.execute( - "UPDATE actors SET repo_rev=$2, repo_cid=$3 WHERE did=$1", - &[&did, &rev, &cid_digest], - ) + // Use consolidated ActorUpdate API for repo state update + use crate::db::operations::{ActorUpdate, ActorUpdateResult, ActorUpdateTarget}; + + let result = ActorUpdate { + target: ActorUpdateTarget::ByDid(did.to_string()), + repo_rev: Some(rev.to_string()), + repo_cid: Some(cid_digest.to_vec()), + ..Default::default() + } + .execute(conn) .await - .wrap_err_with(|| format!("Failed to set repo state for actor {}", did)) + .wrap_err_with(|| format!("Failed to set repo state for actor {}", did))?; + + match result { + ActorUpdateResult::Count(n) => Ok(n), + _ => unreachable!("ActorUpdate with Count returning should return Count"), + } } pub async fn actor_get_statuses( diff --git a/consumer/src/db/operations/actor.rs b/consumer/src/db/operations/actor.rs index 033615ba..9ecafbf4 100644 --- a/consumer/src/db/operations/actor.rs +++ b/consumer/src/db/operations/actor.rs @@ -57,24 +57,22 @@ pub async fn profile_upsert( } pub async fn profile_delete(conn: &C, actor_id: i32) -> Result { - // SCHEMA CHANGE: profiles table dropped, set actors.profile_* columns to NULL - conn.execute( - "UPDATE actors SET - profile_cid = NULL, - profile_created_at = NULL, - profile_avatar_cid = NULL, - profile_banner_cid = NULL, - profile_display_name = NULL, - profile_description = NULL, - profile_pinned_post_rkey = NULL, - profile_joined_sp_id = NULL, - profile_pronouns = NULL, - profile_website = NULL - WHERE id = $1", - &[&actor_id], - ) + // Use consolidated ActorUpdate API for profile deletion + use super::actor_update::{ActorUpdate, ActorUpdateResult, ActorUpdateTarget, ProfileOp}; + + let result = ActorUpdate { + target: ActorUpdateTarget::ById(actor_id), + profile: Some(ProfileOp::Clear), + ..Default::default() + } + .execute(conn) .await - .wrap_err_with(|| format!("Failed to delete profile for actor_id {}", actor_id)) + .wrap_err_with(|| format!("Failed to delete profile for actor_id {}", actor_id))?; + + match result { + ActorUpdateResult::Count(n) => Ok(n), + _ => unreachable!("ActorUpdate with Count returning should return Count"), + } } pub async fn status_upsert( @@ -117,22 +115,22 @@ pub async fn status_upsert( } pub async fn status_delete(conn: &C, actor_id: i32) -> Result { - // SCHEMA CHANGE: statuses table dropped, set actors.status_* columns to NULL - conn.execute( - "UPDATE actors SET - status_cid = NULL, - status_created_at = NULL, - status_type = NULL, - status_duration = NULL, - status_embed_post_actor_id = NULL, - status_embed_post_rkey = NULL, - status_thumb_mime_type = NULL, - status_thumb_cid = NULL - WHERE id = $1", - &[&actor_id], - ) + // Use consolidated ActorUpdate API for status deletion + use super::actor_update::{ActorUpdate, ActorUpdateResult, ActorUpdateTarget, StatusOp}; + + let result = ActorUpdate { + target: ActorUpdateTarget::ById(actor_id), + status: Some(StatusOp::Clear), + ..Default::default() + } + .execute(conn) .await - .wrap_err_with(|| format!("Failed to delete status for actor_id {}", actor_id)) + .wrap_err_with(|| format!("Failed to delete status for actor_id {}", actor_id))?; + + match result { + ActorUpdateResult::Count(n) => Ok(n), + _ => unreachable!("ActorUpdate with Count returning should return Count"), + } } pub async fn chat_decl_upsert( @@ -140,30 +138,43 @@ pub async fn chat_decl_upsert( actor_id: i32, rec: ChatBskyActorDeclaration, ) -> Result { - // SCHEMA CHANGE: chat_decls table dropped, now UPDATE actors.chat_* columns - // No advisory lock needed - simple UPDATE with PostgreSQL row-level locking - conn.execute( - "UPDATE actors SET - chat_allow_incoming = $2::text::chat_allow_incoming, - chat_created_at = NOW() - WHERE id = $1", - &[&actor_id, &rec.allow_incoming.to_string()], - ) + // Use consolidated ActorUpdate API for chat declaration upsert + use super::actor_update::{ActorUpdate, ActorUpdateResult, ActorUpdateTarget, ChatDeclOp}; + + let result = ActorUpdate { + target: ActorUpdateTarget::ById(actor_id), + chat_decl: Some(ChatDeclOp::Set { + allow_incoming: rec.allow_incoming.to_string(), + }), + ..Default::default() + } + .execute(conn) .await - .wrap_err_with(|| format!("Failed to upsert chat declaration for actor_id {}", actor_id)) + .wrap_err_with(|| format!("Failed to upsert chat declaration for actor_id {}", actor_id))?; + + match result { + ActorUpdateResult::Count(n) => Ok(n), + _ => unreachable!("ActorUpdate with Count returning should return Count"), + } } pub async fn chat_decl_delete(conn: &C, actor_id: i32) -> Result { - // SCHEMA CHANGE: chat_decls table dropped, set actors.chat_* columns to NULL - conn.execute( - "UPDATE actors SET - chat_allow_incoming = NULL, - chat_created_at = NULL - WHERE id = $1", - &[&actor_id], - ) + // Use consolidated ActorUpdate API for chat declaration deletion + use super::actor_update::{ActorUpdate, ActorUpdateResult, ActorUpdateTarget, ChatDeclOp}; + + let result = ActorUpdate { + target: ActorUpdateTarget::ById(actor_id), + chat_decl: Some(ChatDeclOp::Clear), + ..Default::default() + } + .execute(conn) .await - .wrap_err_with(|| format!("Failed to delete chat declaration for actor_id {}", actor_id)) + .wrap_err_with(|| format!("Failed to delete chat declaration for actor_id {}", actor_id))?; + + match result { + ActorUpdateResult::Count(n) => Ok(n), + _ => unreachable!("ActorUpdate with Count returning should return Count"), + } } pub async fn notif_decl_upsert( @@ -171,38 +182,51 @@ pub async fn notif_decl_upsert( actor_id: i32, rec: AppBskyNotificationDeclaration, ) -> Result { - // SCHEMA CHANGE: notif_decl table dropped, now UPDATE actors.notif_decl_* columns - // No advisory lock needed - simple UPDATE with PostgreSQL row-level locking - conn.execute( - "UPDATE actors SET - notif_decl_allow_subscriptions = $2::text::notif_allow_subscriptions, - notif_decl_created_at = NOW() - WHERE id = $1", - &[&actor_id, &rec.allow_subscriptions.to_string()], - ) + // Use consolidated ActorUpdate API for notification declaration upsert + use super::actor_update::{ActorUpdate, ActorUpdateResult, ActorUpdateTarget, NotifDeclOp}; + + let result = ActorUpdate { + target: ActorUpdateTarget::ById(actor_id), + notif_decl: Some(NotifDeclOp::Set { + allow_subscriptions: rec.allow_subscriptions.to_string(), + }), + ..Default::default() + } + .execute(conn) .await .wrap_err_with(|| { format!( "Failed to upsert notification declaration for actor_id {}", actor_id ) - }) + })?; + + match result { + ActorUpdateResult::Count(n) => Ok(n), + _ => unreachable!("ActorUpdate with Count returning should return Count"), + } } pub async fn notif_decl_delete(conn: &C, actor_id: i32) -> Result { - // SCHEMA CHANGE: notif_decl table dropped, set actors.notif_decl_* columns to NULL - conn.execute( - "UPDATE actors SET - notif_decl_allow_subscriptions = NULL, - notif_decl_created_at = NULL - WHERE id = $1", - &[&actor_id], - ) + // Use consolidated ActorUpdate API for notification declaration deletion + use super::actor_update::{ActorUpdate, ActorUpdateResult, ActorUpdateTarget, NotifDeclOp}; + + let result = ActorUpdate { + target: ActorUpdateTarget::ById(actor_id), + notif_decl: Some(NotifDeclOp::Clear), + ..Default::default() + } + .execute(conn) .await .wrap_err_with(|| { format!( "Failed to delete notification declaration for actor_id {}", actor_id ) - }) + })?; + + match result { + ActorUpdateResult::Count(n) => Ok(n), + _ => unreachable!("ActorUpdate with Count returning should return Count"), + } } diff --git a/consumer/src/db/operations/actor_update.rs b/consumer/src/db/operations/actor_update.rs new file mode 100644 index 00000000..f4f4ac9e --- /dev/null +++ b/consumer/src/db/operations/actor_update.rs @@ -0,0 +1,647 @@ +//! Consolidated actor UPDATE operations +//! +//! This module provides a single, unified interface for all actor table updates. +//! All UPDATE queries go through the ActorUpdate struct, which dynamically builds +//! SQL based on which fields are provided. +//! +//! **Design Principle: ONE code path for all updates** +//! - Reduces maintenance overhead +//! - Makes SQL easier to test and audit +//! - Prevents duplicate logic +//! +//! # Examples +//! +//! ```ignore +//! // Update profile +//! ActorUpdate { +//! target: ActorUpdateTarget::ById(actor_id), +//! profile: Some(ProfileOp::Set { ... }), +//! ..Default::default() +//! }.execute(conn).await?; +//! +//! // Delete profile (set to NULL) +//! ActorUpdate { +//! target: ActorUpdateTarget::ById(actor_id), +//! profile: Some(ProfileOp::Clear), +//! ..Default::default() +//! }.execute(conn).await?; +//! +//! // Batch update counts with CASE expressions +//! ActorUpdate { +//! target: ActorUpdateTarget::Batch { ids: vec![1, 2, 3] }, +//! count_deltas: Some(HashMap::from([ +//! (1, CountDeltas { followers_count: Some(5), .. }), +//! (2, CountDeltas { posts_count: Some(1), .. }), +//! ])), +//! ..Default::default() +//! }.execute(conn).await?; +//! ``` + +use crate::Result; +use deadpool_postgres::GenericClient; +use parakeet_db::types::{ActorStatus, ActorSyncState}; +use std::collections::HashMap; + +/// Target actors to update +#[derive(Debug, Clone)] +pub enum ActorUpdateTarget { + /// Single actor by ID + ById(i32), + + /// Single actor by DID (will be resolved to ID internally) + ByDid(String), + + /// Batch of actors by IDs + Batch { ids: Vec }, +} + +/// Profile field operation +#[derive(Debug, Clone)] +pub enum ProfileOp { + /// Set profile fields to specific values + Set { + cid: Vec, + created_at: chrono::DateTime, + avatar_cid: Option>, + banner_cid: Option>, + display_name: Option, + description: Option, + pinned_post_rkey: Option, + joined_sp_id: Option, + pronouns: Option, + website: Option, + }, + + /// Clear all profile fields (set to NULL) + Clear, +} + +/// Status field operation +#[derive(Debug, Clone)] +pub enum StatusOp { + /// Set status fields to specific values + Set { + cid: Vec, + created_at: Option>, + status_type: String, + duration_minutes: Option, + embed_post_actor_id: Option, + embed_post_rkey: Option, + thumb_mime_type: Option, + thumb_cid: Option>, + }, + + /// Clear all status fields (set to NULL) + Clear, +} + +/// Chat declaration operation +#[derive(Debug, Clone)] +pub enum ChatDeclOp { + /// Set chat fields + Set { allow_incoming: String }, + + /// Clear chat fields (set to NULL) + Clear, +} + +/// Notification declaration operation +#[derive(Debug, Clone)] +pub enum NotifDeclOp { + /// Set notification declaration fields + Set { allow_subscriptions: String }, + + /// Clear notification declaration fields (set to NULL) + Clear, +} + +/// Count deltas for batch updates +#[derive(Debug, Clone, Default)] +pub struct ActorCountDeltas { + pub followers_count: Option, + pub following_count: Option, + pub posts_count: Option, + pub lists_count: Option, + pub feeds_count: Option, + pub starterpacks_count: Option, +} + +/// What to return from UPDATE +#[derive(Debug, Clone)] +pub enum ActorUpdateReturning { + /// No RETURNING clause, just row count + Count, + + /// Return actor IDs + Ids, +} + +/// Result from ActorUpdate execution +#[derive(Debug)] +pub enum ActorUpdateResult { + /// Number of rows updated + Count(u64), + + /// List of actor IDs updated + Ids(Vec), +} + +/// Main update parameters +#[derive(Debug, Clone)] +pub struct ActorUpdate { + /// Which actors to update + pub target: ActorUpdateTarget, + + /// Field group operations (all optional) + pub profile: Option, + pub status: Option, + pub chat_decl: Option, + pub notif_decl: Option, + + /// Core actor fields (all optional) + pub actor_status: Option, + pub handle: Option, + pub sync_state: Option, + pub sync_state_upgrade_only: bool, // If true, use CASE to prevent downgrades + pub account_created_at: Option>, + pub account_created_at_coalesce: bool, // If true, use COALESCE (don't overwrite existing) + pub last_indexed: Option>, + pub repo_rev: Option, + pub repo_cid: Option>, + + /// Batch count deltas (for batch updates with CASE expressions) + pub count_deltas: Option>, + + /// What to return + pub returning: ActorUpdateReturning, +} + +impl Default for ActorUpdate { + fn default() -> Self { + Self { + target: ActorUpdateTarget::ById(0), + profile: None, + status: None, + chat_decl: None, + notif_decl: None, + actor_status: None, + handle: None, + sync_state: None, + sync_state_upgrade_only: false, + account_created_at: None, + account_created_at_coalesce: false, + last_indexed: None, + repo_rev: None, + repo_cid: None, + count_deltas: None, + returning: ActorUpdateReturning::Count, + } + } +} + +impl ActorUpdate { + /// Execute the update operation + pub async fn execute(self, conn: &C) -> Result { + // Resolve target to WHERE clause + let (where_clause, where_params) = self.build_where(conn).await?; + + // Build SET clause + let (set_clause, mut set_params) = self.build_set()?; + + if set_clause.is_empty() { + return Err(eyre::eyre!("No fields to update")); + } + + // Build RETURNING clause + let returning_clause = self.build_returning(); + + // Renumber WHERE clause parameters to start after SET parameters + let num_set_params = set_params.len(); + let num_where_params = where_params.len(); + let renumbered_where = renumber_params(&where_clause, num_where_params, num_set_params); + + // Combine into final SQL + let sql = if returning_clause.is_empty() { + format!("UPDATE actors SET {} WHERE {}", set_clause, renumbered_where) + } else { + format!( + "UPDATE actors SET {} WHERE {} {}", + set_clause, renumbered_where, returning_clause + ) + }; + + // Combine params: SET params first, then WHERE params + set_params.extend(where_params); + let all_params: Vec<&(dyn tokio_postgres::types::ToSql + Sync)> = + set_params.iter().map(|p| &**p as &(dyn tokio_postgres::types::ToSql + Sync)).collect(); + + // Execute based on returning type + match self.returning { + ActorUpdateReturning::Count => { + let rows = conn.execute(&sql, &all_params[..]).await?; + Ok(ActorUpdateResult::Count(rows)) + } + ActorUpdateReturning::Ids => { + let rows = conn.query(&sql, &all_params[..]).await?; + Ok(ActorUpdateResult::Ids( + rows.iter().map(|r| r.get(0)).collect(), + )) + } + } + } + + /// Build WHERE clause with parameters + async fn build_where( + &self, + conn: &C, + ) -> Result<(String, Vec>)> { + match &self.target { + ActorUpdateTarget::ById(actor_id) => { + let clause = "id = $1".to_string(); + let params: Vec> = + vec![Box::new(*actor_id)]; + Ok((clause, params)) + } + ActorUpdateTarget::ByDid(did) => { + // Resolve DID to ID first + let actor_id = super::super::actor_id_from_did(conn, did) + .await? + .ok_or_else(|| eyre::eyre!("Actor not found for DID: {}", did))?; + + let clause = "id = $1".to_string(); + let params: Vec> = + vec![Box::new(actor_id)]; + Ok((clause, params)) + } + ActorUpdateTarget::Batch { ids } => { + if ids.is_empty() { + return Ok(("FALSE".to_string(), vec![])); + } + let id_list: Vec = ids.iter().map(|id| id.to_string()).collect(); + let clause = format!("id IN ({})", id_list.join(", ")); + Ok((clause, vec![])) + } + } + } + + /// Build SET clause with parameters (params start at $1) + fn build_set(&self) -> Result<(String, Vec>)> { + let mut clauses = Vec::new(); + let mut params: Vec> = Vec::new(); + let mut param_idx = 1; + + // Profile operations + if let Some(profile_op) = &self.profile { + match profile_op { + ProfileOp::Set { + cid, + created_at, + avatar_cid, + banner_cid, + display_name, + description, + pinned_post_rkey, + joined_sp_id, + pronouns, + website, + } => { + clauses.push(format!("profile_cid = ${}", param_idx)); + params.push(Box::new(cid.clone())); + param_idx += 1; + + clauses.push(format!("profile_created_at = ${}", param_idx)); + params.push(Box::new(*created_at)); + param_idx += 1; + + clauses.push(format!("profile_avatar_cid = ${}", param_idx)); + params.push(Box::new(avatar_cid.clone())); + param_idx += 1; + + clauses.push(format!("profile_banner_cid = ${}", param_idx)); + params.push(Box::new(banner_cid.clone())); + param_idx += 1; + + clauses.push(format!("profile_display_name = ${}", param_idx)); + params.push(Box::new(display_name.clone())); + param_idx += 1; + + clauses.push(format!("profile_description = ${}", param_idx)); + params.push(Box::new(description.clone())); + param_idx += 1; + + clauses.push(format!("profile_pinned_post_rkey = ${}", param_idx)); + params.push(Box::new(*pinned_post_rkey)); + param_idx += 1; + + clauses.push(format!("profile_joined_sp_id = ${}", param_idx)); + params.push(Box::new(*joined_sp_id)); + param_idx += 1; + + clauses.push(format!("profile_pronouns = ${}", param_idx)); + params.push(Box::new(pronouns.clone())); + param_idx += 1; + + clauses.push(format!("profile_website = ${}", param_idx)); + params.push(Box::new(website.clone())); + param_idx += 1; + } + ProfileOp::Clear => { + clauses.push("profile_cid = NULL".to_string()); + clauses.push("profile_created_at = NULL".to_string()); + clauses.push("profile_avatar_cid = NULL".to_string()); + clauses.push("profile_banner_cid = NULL".to_string()); + clauses.push("profile_display_name = NULL".to_string()); + clauses.push("profile_description = NULL".to_string()); + clauses.push("profile_pinned_post_rkey = NULL".to_string()); + clauses.push("profile_joined_sp_id = NULL".to_string()); + clauses.push("profile_pronouns = NULL".to_string()); + clauses.push("profile_website = NULL".to_string()); + } + } + } + + // Status operations + if let Some(status_op) = &self.status { + match status_op { + StatusOp::Set { + cid, + created_at, + status_type, + duration_minutes, + embed_post_actor_id, + embed_post_rkey, + thumb_mime_type, + thumb_cid, + } => { + clauses.push(format!("status_cid = ${}", param_idx)); + params.push(Box::new(cid.clone())); + param_idx += 1; + + clauses.push(format!("status_created_at = ${}", param_idx)); + params.push(Box::new(*created_at)); + param_idx += 1; + + clauses.push(format!("status_type = ${}::text::status_type", param_idx)); + params.push(Box::new(status_type.clone())); + param_idx += 1; + + clauses.push(format!("status_duration = ${}", param_idx)); + params.push(Box::new(*duration_minutes)); + param_idx += 1; + + clauses.push(format!("status_embed_post_actor_id = ${}", param_idx)); + params.push(Box::new(*embed_post_actor_id)); + param_idx += 1; + + clauses.push(format!("status_embed_post_rkey = ${}", param_idx)); + params.push(Box::new(*embed_post_rkey)); + param_idx += 1; + + clauses.push(format!("status_thumb_mime_type = ${}", param_idx)); + params.push(Box::new(thumb_mime_type.clone())); + param_idx += 1; + + clauses.push(format!("status_thumb_cid = ${}", param_idx)); + params.push(Box::new(thumb_cid.clone())); + param_idx += 1; + } + StatusOp::Clear => { + clauses.push("status_cid = NULL".to_string()); + clauses.push("status_created_at = NULL".to_string()); + clauses.push("status_type = NULL".to_string()); + clauses.push("status_duration = NULL".to_string()); + clauses.push("status_embed_post_actor_id = NULL".to_string()); + clauses.push("status_embed_post_rkey = NULL".to_string()); + clauses.push("status_thumb_mime_type = NULL".to_string()); + clauses.push("status_thumb_cid = NULL".to_string()); + } + } + } + + // Chat declaration operations + if let Some(chat_op) = &self.chat_decl { + match chat_op { + ChatDeclOp::Set { allow_incoming } => { + clauses.push(format!( + "chat_allow_incoming = ${}::text::chat_allow_incoming", + param_idx + )); + params.push(Box::new(allow_incoming.clone())); + param_idx += 1; + + clauses.push("chat_created_at = NOW()".to_string()); + } + ChatDeclOp::Clear => { + clauses.push("chat_allow_incoming = NULL".to_string()); + clauses.push("chat_created_at = NULL".to_string()); + } + } + } + + // Notification declaration operations + if let Some(notif_op) = &self.notif_decl { + match notif_op { + NotifDeclOp::Set { allow_subscriptions } => { + clauses.push(format!( + "notif_decl_allow_subscriptions = ${}::text::notif_allow_subscriptions", + param_idx + )); + params.push(Box::new(allow_subscriptions.clone())); + param_idx += 1; + + clauses.push("notif_decl_created_at = NOW()".to_string()); + } + NotifDeclOp::Clear => { + clauses.push("notif_decl_allow_subscriptions = NULL".to_string()); + clauses.push("notif_decl_created_at = NULL".to_string()); + } + } + } + + // Core actor fields + if let Some(status) = &self.actor_status { + // PostgreSQL custom types need explicit casting + let status_str = match status { + ActorStatus::Active => "active", + ActorStatus::Suspended => "suspended", + ActorStatus::Deleted => "deleted", + ActorStatus::Deactivated => "deactivated", + ActorStatus::Takendown => "takendown", + }; + clauses.push(format!("status = '{}'::actor_status", status_str)); + } + + if let Some(handle) = &self.handle { + clauses.push(format!("handle = ${}", param_idx)); + params.push(Box::new(handle.clone())); + param_idx += 1; + } + + if let Some(sync_state) = &self.sync_state { + if self.sync_state_upgrade_only { + // Use CASE to prevent downgrade from allowlist states to partial + let sync_state_str = match sync_state { + ActorSyncState::Partial => "partial", + ActorSyncState::Dirty => "dirty", + ActorSyncState::Processing => "processing", + ActorSyncState::Synced => "synced", + }; + clauses.push(format!( + "sync_state = CASE \ + WHEN sync_state IN ('synced', 'dirty', 'processing') AND '{}'::actor_sync_state = 'partial' \ + THEN sync_state \ + ELSE '{}'::actor_sync_state \ + END", + sync_state_str, sync_state_str + )); + } else { + let sync_state_str = match sync_state { + ActorSyncState::Partial => "partial", + ActorSyncState::Dirty => "dirty", + ActorSyncState::Processing => "processing", + ActorSyncState::Synced => "synced", + }; + clauses.push(format!("sync_state = '{}'::actor_sync_state", sync_state_str)); + } + } + + if let Some(account_created_at) = &self.account_created_at { + if self.account_created_at_coalesce { + clauses.push(format!("account_created_at = COALESCE(account_created_at, ${})", param_idx)); + } else { + clauses.push(format!("account_created_at = ${}", param_idx)); + } + params.push(Box::new(*account_created_at)); + param_idx += 1; + } + + if let Some(last_indexed) = &self.last_indexed { + clauses.push(format!("last_indexed = ${}", param_idx)); + params.push(Box::new(*last_indexed)); + param_idx += 1; + } + + if let Some(repo_rev) = &self.repo_rev { + clauses.push(format!("repo_rev = ${}", param_idx)); + params.push(Box::new(repo_rev.clone())); + param_idx += 1; + } + + if let Some(repo_cid) = &self.repo_cid { + clauses.push(format!("repo_cid = ${}", param_idx)); + params.push(Box::new(repo_cid.clone())); + param_idx += 1; + } + + // Count deltas (batch updates with CASE expressions) + if let Some(deltas_map) = &self.count_deltas { + // Build CASE expressions for each count field + // Format: followers_count = CASE id WHEN 1 THEN LEAST(2147483647, GREATEST(0, COALESCE(followers_count, 0) + delta)) WHEN ... END + + let mut case_followers = String::new(); + let mut case_following = String::new(); + let mut case_posts = String::new(); + let mut case_lists = String::new(); + let mut case_feeds = String::new(); + let mut case_starterpacks = String::new(); + + for (idx, (actor_id, deltas)) in deltas_map.iter().enumerate() { + if idx > 0 { + case_followers.push(' '); + case_following.push(' '); + case_posts.push(' '); + case_lists.push(' '); + case_feeds.push(' '); + case_starterpacks.push(' '); + } + + // Build CASE clauses with bounds checking (0-2147483647 for integer) + if let Some(delta) = deltas.followers_count { + case_followers.push_str(&format!( + "WHEN {} THEN LEAST(2147483647, GREATEST(0, COALESCE(followers_count, 0) + {}))::integer", + actor_id, delta + )); + } else { + case_followers.push_str(&format!("WHEN {} THEN followers_count", actor_id)); + } + + if let Some(delta) = deltas.following_count { + case_following.push_str(&format!( + "WHEN {} THEN LEAST(2147483647, GREATEST(0, COALESCE(following_count, 0) + {}))::integer", + actor_id, delta + )); + } else { + case_following.push_str(&format!("WHEN {} THEN following_count", actor_id)); + } + + if let Some(delta) = deltas.posts_count { + case_posts.push_str(&format!( + "WHEN {} THEN LEAST(2147483647, GREATEST(0, COALESCE(posts_count, 0) + {}))::integer", + actor_id, delta + )); + } else { + case_posts.push_str(&format!("WHEN {} THEN posts_count", actor_id)); + } + + if let Some(delta) = deltas.lists_count { + case_lists.push_str(&format!( + "WHEN {} THEN LEAST(2147483647, GREATEST(0, COALESCE(lists_count, 0) + {}))::integer", + actor_id, delta + )); + } else { + case_lists.push_str(&format!("WHEN {} THEN lists_count", actor_id)); + } + + if let Some(delta) = deltas.feeds_count { + case_feeds.push_str(&format!( + "WHEN {} THEN LEAST(2147483647, GREATEST(0, COALESCE(feeds_count, 0) + {}))::integer", + actor_id, delta + )); + } else { + case_feeds.push_str(&format!("WHEN {} THEN feeds_count", actor_id)); + } + + if let Some(delta) = deltas.starterpacks_count { + case_starterpacks.push_str(&format!( + "WHEN {} THEN LEAST(2147483647, GREATEST(0, COALESCE(starterpacks_count, 0) + {}))::integer", + actor_id, delta + )); + } else { + case_starterpacks.push_str(&format!("WHEN {} THEN starterpacks_count", actor_id)); + } + } + + // Add SET clauses with CASE expressions + clauses.push(format!("followers_count = CASE id {} END", case_followers)); + clauses.push(format!("following_count = CASE id {} END", case_following)); + clauses.push(format!("posts_count = CASE id {} END", case_posts)); + clauses.push(format!("lists_count = CASE id {} END", case_lists)); + clauses.push(format!("feeds_count = CASE id {} END", case_feeds)); + clauses.push(format!("starterpacks_count = CASE id {} END", case_starterpacks)); + } + + Ok((clauses.join(", "), params)) + } + + /// Build RETURNING clause + fn build_returning(&self) -> String { + match self.returning { + ActorUpdateReturning::Count => String::new(), + ActorUpdateReturning::Ids => "RETURNING id".to_string(), + } + } +} + +/// Renumber SQL parameter placeholders +/// +/// Takes SQL like "field = $1, other = $2" and renumbers to start at `start_idx` +fn renumber_params(sql: &str, num_params: usize, start_idx: usize) -> String { + let mut result = sql.to_string(); + // Renumber from highest to lowest to avoid conflicts + for i in (1..=num_params).rev() { + let old = format!("${}", i); + let new = format!("${}", start_idx + i); + result = result.replace(&old, &new); + } + result +} diff --git a/consumer/src/db/operations/mod.rs b/consumer/src/db/operations/mod.rs index 9b882fa1..894db0d2 100644 --- a/consumer/src/db/operations/mod.rs +++ b/consumer/src/db/operations/mod.rs @@ -1,4 +1,5 @@ pub mod actor; +pub mod actor_update; pub mod community; pub mod feed; pub mod graph; @@ -7,6 +8,7 @@ pub mod starter_pack; // Re-export all public functions pub use actor::*; +pub use actor_update::*; pub use community::*; pub use feed::*; pub use graph::*;