| 234 | } |
| 235 | |
| 236 | fn apply_updates( |
| 237 | item: &QueueItem, |
| 238 | updates: &serde_json::Map<String, Value>, |
| 239 | ) -> Result<(), String> { |
| 240 | if updates.is_empty() { |
| 241 | return Ok(()); |
| 242 | } |
| 243 | |
| 244 | let memory_update = updates.get("__memory"); |
| 245 | let column_updates: serde_json::Map<String, Value> = updates |
| 246 | .iter() |
| 247 | .filter(|(k, _)| k.as_str() != "__memory") |
| 248 | .map(|(k, v)| (k.clone(), v.clone())) |
| 249 | .collect(); |
| 250 | |
| 251 | let table_name = Spi::get_one::<String>(&format!( |
| 252 | "SELECT relname::text FROM pg_class WHERE oid = {}", |
| 253 | item.table_oid.as_u32() |
| 254 | )) |
| 255 | .expect("SPI failed") |
| 256 | .expect("table not found"); |
| 257 | |
| 258 | let pk_columns = Spi::get_one::<Vec<String>>(&format!( |
| 259 | "SELECT pk_columns FROM claw.watched WHERE table_oid = {}", |
| 260 | item.table_oid.as_u32() |
| 261 | )) |
| 262 | .expect("SPI failed") |
| 263 | .expect("watched table not found"); |
| 264 | |
| 265 | if !column_updates.is_empty() { |
| 266 | let set_parts: Vec<String> = column_updates |
| 267 | .iter() |
| 268 | .map(|(col, val)| { |
| 269 | let val_str = match val { |
| 270 | Value::String(s) => format!("'{}'", s.replace('\'', "''")), |
| 271 | Value::Null => "NULL".to_string(), |
| 272 | other => format!("'{}'", other.to_string().replace('\'', "''")), |
| 273 | }; |
| 274 | format!("{} = {}", col, val_str) |
| 275 | }) |
| 276 | .collect(); |
| 277 | |
| 278 | let pk_values: Vec<&str> = item.pk_value.split('|').collect(); |
| 279 | let where_parts: Vec<String> = pk_columns |
| 280 | .iter() |
| 281 | .zip(pk_values.iter()) |
| 282 | .map(|(col, val)| format!("{} = '{}'", col, val.replace('\'', "''"))) |
| 283 | .collect(); |
| 284 | |
| 285 | let sql = format!( |
| 286 | "SET pgclaw.skip_trigger = 'true'; UPDATE {} SET {} WHERE {}; SET pgclaw.skip_trigger = 'false'", |
| 287 | table_name, |
| 288 | set_parts.join(", "), |
| 289 | where_parts.join(" AND ") |
| 290 | ); |
| 291 | |
| 292 | Spi::run(&sql).map_err(|e| format!("update failed: {}", e))?; |
| 293 | } |