ingest.rs - lib - Core libraries for Radroots

ingest.rs (168084B)
      1 #[cfg(not(feature = "std"))]
      2 use alloc::format;
      3 #[cfg(not(feature = "std"))]
      4 use alloc::{
      5     string::{String, ToString},
      6     vec::Vec,
      7 };
      8 
      9 #[cfg(feature = "std")]
     10 use base64::Engine;
     11 #[cfg(feature = "std")]
     12 use base64::engine::general_purpose::URL_SAFE_NO_PAD;
     13 
     14 use radroots_core::Decimal;
     15 use radroots_event::contract::EventClass;
     16 #[cfg(test)]
     17 use radroots_event::envelope::EventEnvelopeParts;
     18 use radroots_event::envelope::event_head::{
     19     CurrentEventHead, EventHeadCandidateResult, EventHeadCoordinate,
     20     EventHeadDecision as ProtocolEventHeadDecision, event_head_candidate_for_class,
     21     event_head_candidate_for_event, select_event_head,
     22 };
     23 use radroots_event::envelope::kind::{
     24     KIND_CALENDAR, KIND_CLASSIFIED_LISTING, KIND_FARM, KIND_PLOT, KIND_PROFILE,
     25     is_nip51_list_set_kind,
     26 };
     27 use radroots_event::id::EventId;
     28 use radroots_event::listing::operational::{
     29     OperationalListing, OperationalListingAvailability, OperationalListingBin,
     30     OperationalListingStatus,
     31 };
     32 use radroots_event::profile::{
     33     ProfileType, RADROOTS_PROFILE_TYPE_TAG_KEY, radroots_profile_type_from_tag_value,
     34 };
     35 use radroots_event::{
     36     envelope::EventEnvelope,
     37     listing::classified::{ClassifiedListingPartition, classify_classified_listing_tags},
     38 };
     39 use radroots_event_codec::admission::profile::admit_verified_profile_event;
     40 use radroots_event_codec::decode::farm as farm_decode;
     41 use radroots_event_codec::decode::food_availability::{
     42     RadrootsFoodAvailabilityProjectionOutcome, project_verified_food_availability_event,
     43 };
     44 use radroots_event_codec::decode::list_set as list_set_decode;
     45 use radroots_event_codec::decode::operational_listing as listing_decode;
     46 use radroots_event_codec::decode::plot as plot_decode;
     47 use radroots_event_codec::verify::{RadrootsSignatureVerifiedEvent, verify_nip01_event};
     48 use radroots_replica_schema::ReplicaSchemaError;
     49 use radroots_replica_schema::farm::{
     50     FarmQueryBindValues, IFarmFields, IFarmFieldsFilter, IFarmFieldsPartial, IFarmFindMany,
     51     IFarmUpdate,
     52 };
     53 #[cfg(test)]
     54 use radroots_replica_schema::farm_gcs_location::{
     55     FarmGcsLocationQueryBindValues, IFarmGcsLocationDelete, IFarmGcsLocationFieldsFilter,
     56     IFarmGcsLocationFindMany, IFarmGcsLocationFindOneArgs,
     57 };
     58 use radroots_replica_schema::farm_member::{
     59     FarmMemberQueryBindValues, IFarmMemberDelete, IFarmMemberFields, IFarmMemberFieldsFilter,
     60     IFarmMemberFindMany, IFarmMemberFindOneArgs,
     61 };
     62 use radroots_replica_schema::farm_member_claim::{
     63     FarmMemberClaimQueryBindValues, IFarmMemberClaimDelete, IFarmMemberClaimFields,
     64     IFarmMemberClaimFieldsFilter, IFarmMemberClaimFindMany, IFarmMemberClaimFindOneArgs,
     65 };
     66 use radroots_replica_schema::farm_tag::{
     67     FarmTagQueryBindValues, IFarmTagDelete, IFarmTagFields, IFarmTagFieldsFilter, IFarmTagFindMany,
     68     IFarmTagFindOneArgs,
     69 };
     70 use radroots_replica_schema::gcs_location::IGcsLocationFields;
     71 use radroots_replica_schema::nostr_event_head::{
     72     INostrEventHeadFields, INostrEventHeadFieldsPartial, INostrEventHeadFindOne,
     73     INostrEventHeadFindOneArgs, INostrEventHeadUpdate, NostrEventHead,
     74     NostrEventHeadQueryBindValues,
     75 };
     76 use radroots_replica_schema::nostr_profile::{
     77     INostrProfileFields, INostrProfileFieldsPartial, INostrProfileFindOne,
     78     INostrProfileFindOneArgs, INostrProfileUpdate, NostrProfileQueryBindValues,
     79 };
     80 use radroots_replica_schema::plot::{
     81     IPlotFields, IPlotFieldsFilter, IPlotFieldsPartial, IPlotFindMany, IPlotUpdate,
     82     PlotQueryBindValues,
     83 };
     84 use radroots_replica_schema::plot_gcs_location::{
     85     IPlotGcsLocationDelete, IPlotGcsLocationFields, IPlotGcsLocationFieldsFilter,
     86     IPlotGcsLocationFindMany, IPlotGcsLocationFindOneArgs, PlotGcsLocationQueryBindValues,
     87 };
     88 use radroots_replica_schema::plot_tag::{
     89     IPlotTagDelete, IPlotTagFields, IPlotTagFieldsFilter, IPlotTagFindMany, IPlotTagFindOneArgs,
     90     PlotTagQueryBindValues,
     91 };
     92 use radroots_replica_schema::trade_product::{
     93     ITradeProductFields, ITradeProductFieldsFilter, ITradeProductFieldsPartial,
     94     ITradeProductFindMany, ITradeProductFindOne, ITradeProductFindOneArgs, ITradeProductUpdate,
     95     TradeProductQueryBindValues,
     96 };
     97 #[cfg(test)]
     98 use radroots_replica_store::farm_gcs_location;
     99 use radroots_replica_store::{
    100     farm, farm_member, farm_member_claim, farm_tag, gcs_location, nostr_event_head, nostr_profile,
    101     plot, plot_gcs_location, plot_tag, trade_product,
    102 };
    103 use radroots_sql_core::SqlExecutor;
    104 use radroots_sql_core::error::SqlError;
    105 use serde_json::{Value, json};
    106 
    107 use crate::error::RadrootsReplicaEventsError;
    108 use crate::event_head::{event_content_hash, event_head_key};
    109 const ROLE_PRIMARY: &str = "primary";
    110 const ROLE_MEMBER: &str = "member";
    111 const ROLE_OWNER: &str = "owner";
    112 const ROLE_WORKER: &str = "worker";
    113 
    114 #[cfg(test)]
    115 pub(crate) mod failpoints {
    116     use std::cell::Cell;
    117 
    118     thread_local! {
    119         static FORCE_GCS_POINT_SERIALIZE_ERROR: Cell<bool> = const { Cell::new(false) };
    120         static FORCE_GCS_POLYGON_SERIALIZE_ERROR: Cell<bool> = const { Cell::new(false) };
    121     }
    122 
    123     pub(crate) fn set_gcs_point_serialize_error() {
    124         FORCE_GCS_POINT_SERIALIZE_ERROR.with(|flag| flag.set(true));
    125     }
    126 
    127     pub(crate) fn take_gcs_point_serialize_error() -> bool {
    128         FORCE_GCS_POINT_SERIALIZE_ERROR.with(|flag| {
    129             let value = flag.get();
    130             flag.set(false);
    131             value
    132         })
    133     }
    134 
    135     pub(crate) fn set_gcs_polygon_serialize_error() {
    136         FORCE_GCS_POLYGON_SERIALIZE_ERROR.with(|flag| flag.set(true));
    137     }
    138 
    139     pub(crate) fn take_gcs_polygon_serialize_error() -> bool {
    140         FORCE_GCS_POLYGON_SERIALIZE_ERROR.with(|flag| {
    141             let value = flag.get();
    142             flag.set(false);
    143             value
    144         })
    145     }
    146 }
    147 
    148 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
    149 pub enum RadrootsReplicaIngestOutcome {
    150     /// The selected event updated its supported replica projection and raw head.
    151     Applied,
    152     /// The selected raw head belongs to a valid profile this replica excludes.
    153     Excluded,
    154     /// The selected raw head is invalid or ambiguous for its declared profile.
    155     Rejected,
    156     /// The event did not win NIP-01 replacement ordering.
    157     Skipped,
    158 }
    159 
    160 pub trait RadrootsReplicaIdFactory {
    161     fn new_d_tag(&self) -> String;
    162 }
    163 
    164 #[cfg(feature = "std")]
    165 pub struct RadrootsReplicaDefaultIdFactory;
    166 
    167 #[cfg(feature = "std")]
    168 impl RadrootsReplicaIdFactory for RadrootsReplicaDefaultIdFactory {
    169     fn new_d_tag(&self) -> String {
    170         let uuid = uuid::Uuid::now_v7();
    171         let bytes = uuid.as_bytes();
    172         URL_SAFE_NO_PAD.encode(bytes)
    173     }
    174 }
    175 
    176 #[cfg(feature = "std")]
    177 /// Verifies and ingests an envelope through its supported replica projection.
    178 pub fn radroots_replica_ingest_event(
    179     exec: &dyn SqlExecutor,
    180     event: &EventEnvelope,
    181 ) -> Result<RadrootsReplicaIngestOutcome, RadrootsReplicaEventsError> {
    182     radroots_replica_ingest_event_with_factory(exec, event, &RadrootsReplicaDefaultIdFactory)
    183 }
    184 
    185 /// Verifies and ingests an envelope with an explicit replica ID source.
    186 pub fn radroots_replica_ingest_event_with_factory(
    187     exec: &dyn SqlExecutor,
    188     event: &EventEnvelope,
    189     factory: &dyn RadrootsReplicaIdFactory,
    190 ) -> Result<RadrootsReplicaIngestOutcome, RadrootsReplicaEventsError> {
    191     let verified_event = match event.kind_u32() {
    192         KIND_PROFILE | KIND_CLASSIFIED_LISTING => Some(verify_nip01_event(event.clone())?),
    193         _ => None,
    194     };
    195 
    196     if let Err(err) = exec.begin() {
    197         return Err(RadrootsReplicaEventsError::from(ReplicaSchemaError::from(
    198             err,
    199         )));
    200     }
    201 
    202     match ingest_event_inner(exec, event, factory, verified_event.as_ref()) {
    203         Ok(outcome) => {
    204             if let Err(err) = exec.commit() {
    205                 return Err(RadrootsReplicaEventsError::from(ReplicaSchemaError::from(
    206                     err,
    207                 )));
    208             }
    209             Ok(outcome)
    210         }
    211         Err(err) => {
    212             let _ = exec.rollback();
    213             Err(err)
    214         }
    215     }
    216 }
    217 
    218 fn ingest_event_inner(
    219     exec: &dyn SqlExecutor,
    220     event: &EventEnvelope,
    221     factory: &dyn RadrootsReplicaIdFactory,
    222     verified_event: Option<&RadrootsSignatureVerifiedEvent>,
    223 ) -> Result<RadrootsReplicaIngestOutcome, RadrootsReplicaEventsError> {
    224     match event.kind_u32() {
    225         KIND_PROFILE => {
    226             let verified_event = verified_event.ok_or_else(|| {
    227                 RadrootsReplicaEventsError::InvalidData(
    228                     "profile verification invariant missing".to_string(),
    229                 )
    230             })?;
    231             ingest_profile_event(exec, verified_event)
    232         }
    233         KIND_FARM => ingest_farm_event(exec, event, factory),
    234         KIND_PLOT => ingest_plot_event(exec, event, factory),
    235         KIND_CLASSIFIED_LISTING => {
    236             let verified_event = verified_event.ok_or_else(|| {
    237                 RadrootsReplicaEventsError::InvalidData(
    238                     "classified listing verification invariant missing".to_string(),
    239                 )
    240             })?;
    241             ingest_listing_event(exec, verified_event)
    242         }
    243         kind if is_nip51_list_set_kind(kind) && kind != KIND_CALENDAR => {
    244             ingest_list_set_event(exec, event)
    245         }
    246         _ => Err(RadrootsReplicaEventsError::InvalidData(format!(
    247             "unsupported kind {}",
    248             event.kind_u32()
    249         ))),
    250     }
    251 }
    252 
    253 fn ingest_profile_event(
    254     exec: &dyn SqlExecutor,
    255     verified_event: &RadrootsSignatureVerifiedEvent,
    256 ) -> Result<RadrootsReplicaIngestOutcome, RadrootsReplicaEventsError> {
    257     let admitted = admit_verified_profile_event(verified_event.clone())
    258         .map_err(|error| RadrootsReplicaEventsError::InvalidData(error.to_string()))?;
    259     let event = admitted.event();
    260     let metadata = admitted.metadata();
    261     let profile_type = match profile_type_from_event(event) {
    262         Some(profile_type) => profile_type,
    263         None => {
    264             return Err(RadrootsReplicaEventsError::InvalidData(
    265                 "profile_type required".to_string(),
    266             ));
    267         }
    268     };
    269 
    270     let decision = event_head_decision(exec, event)?;
    271     if !decision.apply {
    272         return Ok(RadrootsReplicaIngestOutcome::Skipped);
    273     }
    274 
    275     let profile_type = match profile_type {
    276         radroots_event::profile::ProfileType::Individual => "individual",
    277         radroots_event::profile::ProfileType::Farm => "farm",
    278         radroots_event::profile::ProfileType::Coop => "coop",
    279         radroots_event::profile::ProfileType::Any => "any",
    280         radroots_event::profile::ProfileType::Radrootsd => "radrootsd",
    281     };
    282 
    283     let existing_result = nostr_profile::find_one(
    284         exec,
    285         &INostrProfileFindOne::On(INostrProfileFindOneArgs {
    286             on: NostrProfileQueryBindValues::PublicKey {
    287                 public_key: event.author().to_hex().to_owned(),
    288             },
    289         }),
    290     );
    291     let existing = existing_result?.result;
    292 
    293     match existing {
    294         Some(profile) => {
    295             let fields = INostrProfileFieldsPartial {
    296                 public_key: None,
    297                 profile_type: Some(Value::from(profile_type)),
    298                 name: Some(Value::from(required_profile_name(metadata)?)),
    299                 display_name: to_value_opt(metadata.display_name().map(str::to_owned)),
    300                 about: to_value_opt(metadata.about().map(str::to_owned)),
    301                 website: to_value_opt(profile_string_field(metadata, "website")),
    302                 picture: to_value_opt(metadata.picture().map(|value| value.as_str().to_owned())),
    303                 banner: to_value_opt(metadata.banner().map(|value| value.as_str().to_owned())),
    304                 nip05: to_value_opt(metadata.nip05().map(|value| value.as_str().to_owned())),
    305                 lud06: to_value_opt(profile_string_field(metadata, "lud06")),
    306                 lud16: to_value_opt(profile_string_field(metadata, "lud16")),
    307             };
    308             let update_result = nostr_profile::update(
    309                 exec,
    310                 &INostrProfileUpdate {
    311                     on: NostrProfileQueryBindValues::Id { id: profile.id },
    312                     fields,
    313                 },
    314             );
    315             let _updated = update_result?;
    316         }
    317         None => {
    318             let fields = INostrProfileFields {
    319                 public_key: event.author().to_hex().to_owned(),
    320                 profile_type: profile_type.to_string(),
    321                 name: required_profile_name(metadata)?,
    322                 display_name: metadata.display_name().map(str::to_owned),
    323                 about: metadata.about().map(str::to_owned),
    324                 website: profile_string_field(metadata, "website"),
    325                 picture: metadata.picture().map(|value| value.as_str().to_owned()),
    326                 banner: metadata.banner().map(|value| value.as_str().to_owned()),
    327                 nip05: metadata.nip05().map(|value| value.as_str().to_owned()),
    328                 lud06: profile_string_field(metadata, "lud06"),
    329                 lud16: profile_string_field(metadata, "lud16"),
    330             };
    331             let _ = nostr_profile::create(exec, &fields)?;
    332         }
    333     }
    334 
    335     upsert_event_head(exec, &decision)?;
    336     Ok(RadrootsReplicaIngestOutcome::Applied)
    337 }
    338 
    339 fn profile_type_from_event(event: &EventEnvelope) -> Option<ProfileType> {
    340     event
    341         .tags_as_vec()
    342         .into_iter()
    343         .filter(|tag| {
    344             tag.first()
    345                 .is_some_and(|key| key == RADROOTS_PROFILE_TYPE_TAG_KEY)
    346         })
    347         .filter_map(|tag| tag.get(1).cloned())
    348         .find_map(|value| radroots_profile_type_from_tag_value(&value))
    349 }
    350 
    351 fn required_profile_name(
    352     metadata: &radroots_event_codec::decode::profile::RadrootsInboundProfileMetadata,
    353 ) -> Result<String, RadrootsReplicaEventsError> {
    354     metadata
    355         .name()
    356         .map(str::to_owned)
    357         .ok_or_else(|| RadrootsReplicaEventsError::InvalidData("profile name required".to_string()))
    358 }
    359 
    360 fn profile_string_field(
    361     metadata: &radroots_event_codec::decode::profile::RadrootsInboundProfileMetadata,
    362     field: &'static str,
    363 ) -> Option<String> {
    364     metadata
    365         .raw_fields()
    366         .get(field)
    367         .and_then(Value::as_str)
    368         .map(str::to_owned)
    369 }
    370 
    371 fn ingest_farm_event(
    372     exec: &dyn SqlExecutor,
    373     event: &EventEnvelope,
    374     _factory: &dyn RadrootsReplicaIdFactory,
    375 ) -> Result<RadrootsReplicaIngestOutcome, RadrootsReplicaEventsError> {
    376     let farm =
    377         farm_decode::farm_from_event(event.kind_u32(), &event.tags_as_vec(), event.content())?;
    378     let decision = event_head_decision(exec, event)?;
    379     if !decision.apply {
    380         return Ok(RadrootsReplicaIngestOutcome::Skipped);
    381     }
    382 
    383     let filter = IFarmFieldsFilter {
    384         id: None,
    385         created_at: None,
    386         updated_at: None,
    387         d_tag: Some(farm.d_tag.clone()),
    388         pubkey: Some(event.author().to_hex().to_owned()),
    389         name: None,
    390         about: None,
    391         website: None,
    392         picture: None,
    393         banner: None,
    394         location_primary: None,
    395         location_city: None,
    396         location_region: None,
    397         location_country: None,
    398     };
    399     let existing_result = farm::find_many(
    400         exec,
    401         &IFarmFindMany {
    402             filter: Some(filter),
    403         },
    404     );
    405     let existing = existing_result?;
    406     let location = farm.location.clone();
    407     let (location_primary, location_city, location_region, location_country) =
    408         unpack_farm_location_strings(location.as_ref());
    409     let farm_id = if let Some(row) = existing.results.first() {
    410         let fields = IFarmFieldsPartial {
    411             d_tag: Some(Value::from(farm.d_tag.clone())),
    412             pubkey: Some(Value::from(event.author().to_hex().to_owned())),
    413             name: Some(Value::from(farm.name.clone())),
    414             about: to_value_opt(farm.about.clone()),
    415             website: to_value_opt(farm.website.clone()),
    416             picture: to_value_opt(farm.picture.clone()),
    417             banner: to_value_opt(farm.banner.clone()),
    418             location_primary: to_value_opt(location_primary),
    419             location_city: to_value_opt(location_city),
    420             location_region: to_value_opt(location_region),
    421             location_country: to_value_opt(location_country),
    422         };
    423         let update_result = farm::update(
    424             exec,
    425             &IFarmUpdate {
    426                 on: FarmQueryBindValues::Id { id: row.id.clone() },
    427                 fields,
    428             },
    429         );
    430         let _updated = update_result?;
    431         row.id.clone()
    432     } else {
    433         let fields = IFarmFields {
    434             d_tag: farm.d_tag.clone(),
    435             pubkey: event.author().to_hex().to_owned(),
    436             name: farm.name.clone(),
    437             about: farm.about.clone(),
    438             website: farm.website.clone(),
    439             picture: farm.picture.clone(),
    440             banner: farm.banner.clone(),
    441             location_primary,
    442             location_city,
    443             location_region,
    444             location_country,
    445         };
    446         farm::create(exec, &fields)?.result.id
    447     };
    448 
    449     upsert_farm_tags(exec, &farm_id, farm.tags)?;
    450 
    451     upsert_event_head(exec, &decision)?;
    452     Ok(RadrootsReplicaIngestOutcome::Applied)
    453 }
    454 
    455 fn ingest_plot_event(
    456     exec: &dyn SqlExecutor,
    457     event: &EventEnvelope,
    458     factory: &dyn RadrootsReplicaIdFactory,
    459 ) -> Result<RadrootsReplicaIngestOutcome, RadrootsReplicaEventsError> {
    460     let plot =
    461         plot_decode::plot_from_event(event.kind_u32(), &event.tags_as_vec(), event.content())?;
    462     let decision = event_head_decision(exec, event)?;
    463     if !decision.apply {
    464         return Ok(RadrootsReplicaIngestOutcome::Skipped);
    465     }
    466 
    467     let farm = find_farm_by_ref(exec, &plot.farm.pubkey, &plot.farm.d_tag)?;
    468     let filter = IPlotFieldsFilter {
    469         id: None,
    470         created_at: None,
    471         updated_at: None,
    472         d_tag: Some(plot.d_tag.clone()),
    473         farm_id: Some(farm.id.clone()),
    474         name: None,
    475         about: None,
    476         location_primary: None,
    477         location_city: None,
    478         location_region: None,
    479         location_country: None,
    480     };
    481     let existing_result = plot::find_many(
    482         exec,
    483         &IPlotFindMany {
    484             filter: Some(filter),
    485         },
    486     );
    487     let existing = existing_result?;
    488     let location = plot.location.clone();
    489     let (location_primary, location_city, location_region, location_country) =
    490         unpack_plot_location_strings(location.as_ref());
    491     let plot_id = if let Some(row) = existing.results.first() {
    492         let fields = IPlotFieldsPartial {
    493             d_tag: Some(Value::from(plot.d_tag.clone())),
    494             farm_id: Some(Value::from(farm.id.clone())),
    495             name: Some(Value::from(plot.name.clone())),
    496             about: to_value_opt(plot.about.clone()),
    497             location_primary: to_value_opt(location_primary),
    498             location_city: to_value_opt(location_city),
    499             location_region: to_value_opt(location_region),
    500             location_country: to_value_opt(location_country),
    501         };
    502         let update_result = plot::update(
    503             exec,
    504             &IPlotUpdate {
    505                 on: PlotQueryBindValues::Id { id: row.id.clone() },
    506                 fields,
    507             },
    508         );
    509         let _updated = update_result?;
    510         row.id.clone()
    511     } else {
    512         let fields = IPlotFields {
    513             d_tag: plot.d_tag.clone(),
    514             farm_id: farm.id.clone(),
    515             name: plot.name.clone(),
    516             about: plot.about.clone(),
    517             location_primary,
    518             location_city,
    519             location_region,
    520             location_country,
    521         };
    522         plot::create(exec, &fields)?.result.id
    523     };
    524 
    525     upsert_plot_tags(exec, &plot_id, plot.tags)?;
    526     upsert_plot_location(exec, &plot_id, location, factory)?;
    527 
    528     upsert_event_head(exec, &decision)?;
    529     Ok(RadrootsReplicaIngestOutcome::Applied)
    530 }
    531 
    532 fn ingest_listing_event(
    533     exec: &dyn SqlExecutor,
    534     verified_event: &RadrootsSignatureVerifiedEvent,
    535 ) -> Result<RadrootsReplicaIngestOutcome, RadrootsReplicaEventsError> {
    536     let event = verified_event.event();
    537     let decision = event_head_decision(exec, event)?;
    538     if !decision.apply {
    539         return Ok(RadrootsReplicaIngestOutcome::Skipped);
    540     }
    541 
    542     let partition = classify_classified_listing_tags(event.tags());
    543     if partition == ClassifiedListingPartition::FocusedFoodAvailability {
    544         let outcome = match project_verified_food_availability_event(verified_event) {
    545             Ok(RadrootsFoodAvailabilityProjectionOutcome::Focused(_)) => {
    546                 RadrootsReplicaIngestOutcome::Excluded
    547             }
    548             Ok(RadrootsFoodAvailabilityProjectionOutcome::Excluded(_)) | Err(_) => {
    549                 RadrootsReplicaIngestOutcome::Rejected
    550             }
    551             Ok(_) => RadrootsReplicaIngestOutcome::Rejected,
    552         };
    553         return replace_listing_projection_with_raw_head(exec, &decision, outcome);
    554     }
    555     if partition == ClassifiedListingPartition::GenericNip99 {
    556         return replace_listing_projection_with_raw_head(
    557             exec,
    558             &decision,
    559             RadrootsReplicaIngestOutcome::Excluded,
    560         );
    561     }
    562     if partition == ClassifiedListingPartition::Ambiguous {
    563         return replace_listing_projection_with_raw_head(
    564             exec,
    565             &decision,
    566             RadrootsReplicaIngestOutcome::Rejected,
    567         );
    568     }
    569 
    570     let listing = match listing_decode::operational_listing_from_event(
    571         event.kind_u32(),
    572         &event.tags_as_vec(),
    573         event.content(),
    574     ) {
    575         Ok(listing) => listing,
    576         Err(_) => {
    577             return replace_listing_projection_with_raw_head(
    578                 exec,
    579                 &decision,
    580                 RadrootsReplicaIngestOutcome::Rejected,
    581             );
    582         }
    583     };
    584 
    585     let listing_addr = decision.key.as_str();
    586     if listing_is_orderable(&listing) {
    587         let fields = match trade_product_fields_from_listing(&listing, listing_addr) {
    588             Ok(fields) => fields,
    589             Err(_) => {
    590                 return replace_listing_projection_with_raw_head(
    591                     exec,
    592                     &decision,
    593                     RadrootsReplicaIngestOutcome::Rejected,
    594                 );
    595             }
    596         };
    597         upsert_trade_product_for_listing_addr(exec, listing_addr, fields)?;
    598     } else {
    599         delete_trade_products_for_listing_addr(exec, listing_addr)?;
    600     }
    601 
    602     upsert_event_head(exec, &decision)?;
    603     Ok(RadrootsReplicaIngestOutcome::Applied)
    604 }
    605 
    606 fn replace_listing_projection_with_raw_head(
    607     exec: &dyn SqlExecutor,
    608     decision: &EventHeadDecision,
    609     outcome: RadrootsReplicaIngestOutcome,
    610 ) -> Result<RadrootsReplicaIngestOutcome, RadrootsReplicaEventsError> {
    611     debug_assert!(matches!(
    612         outcome,
    613         RadrootsReplicaIngestOutcome::Excluded | RadrootsReplicaIngestOutcome::Rejected
    614     ));
    615     delete_trade_products_for_listing_addr(exec, &decision.key)?;
    616     upsert_event_head(exec, decision)?;
    617     Ok(outcome)
    618 }
    619 
    620 fn ingest_list_set_event(
    621     exec: &dyn SqlExecutor,
    622     event: &EventEnvelope,
    623 ) -> Result<RadrootsReplicaIngestOutcome, RadrootsReplicaEventsError> {
    624     if event.kind_u32() != radroots_event::envelope::kind::KIND_LIST_SET_GENERIC {
    625         return Ok(RadrootsReplicaIngestOutcome::Skipped);
    626     }
    627     let list_set = list_set_decode::list_set_from_tags(
    628         event.kind_u32(),
    629         event.content().to_owned(),
    630         &event.tags_as_vec(),
    631     )?;
    632 
    633     let metadata_count = usize::from(list_set.title.is_some())
    634         + usize::from(list_set.description.is_some())
    635         + usize::from(list_set.image.is_some());
    636     if metadata_count != 0 {
    637         return Err(RadrootsReplicaEventsError::InvalidData(
    638             "domain:farm list sets must omit metadata".to_string(),
    639         ));
    640     }
    641     if !list_set.content.is_empty() {
    642         return Err(RadrootsReplicaEventsError::InvalidData(
    643             "domain:farm list sets must not include content".to_string(),
    644         ));
    645     }
    646 
    647     let d_tag = list_set.d_tag.clone();
    648 
    649     if d_tag == "member_of.farms" {
    650         ensure_list_set_entries_tag(&list_set, "p", "member_of.farms")?;
    651         let decision = event_head_decision(exec, event)?;
    652         if !decision.apply {
    653             return Ok(RadrootsReplicaIngestOutcome::Skipped);
    654         }
    655         upsert_member_claims(exec, &event.author().to_hex(), &list_set)?;
    656         upsert_event_head(exec, &decision)?;
    657         return Ok(RadrootsReplicaIngestOutcome::Applied);
    658     }
    659 
    660     if let Some((farm_d_tag, role)) = parse_farm_list_set_d_tag(&d_tag) {
    661         if role == ListSetRole::Plots {
    662             ensure_list_set_entries_tag(&list_set, "a", "farm plots")?;
    663             let decision = event_head_decision(exec, event)?;
    664             if !decision.apply {
    665                 return Ok(RadrootsReplicaIngestOutcome::Skipped);
    666             }
    667             upsert_event_head(exec, &decision)?;
    668             return Ok(RadrootsReplicaIngestOutcome::Applied);
    669         }
    670         ensure_list_set_entries_tag(&list_set, "p", "farm members")?;
    671         let decision = event_head_decision(exec, event)?;
    672         if !decision.apply {
    673             return Ok(RadrootsReplicaIngestOutcome::Skipped);
    674         }
    675         let farm = find_farm_by_ref(exec, &event.author().to_hex(), &farm_d_tag)?;
    676         upsert_farm_members(exec, &farm.id, role, &list_set)?;
    677         upsert_event_head(exec, &decision)?;
    678         return Ok(RadrootsReplicaIngestOutcome::Applied);
    679     }
    680 
    681     Err(RadrootsReplicaEventsError::InvalidData(
    682         "unsupported list set d_tag".to_string(),
    683     ))
    684 }
    685 
    686 fn listing_is_orderable(listing: &OperationalListing) -> bool {
    687     match listing.availability.as_ref() {
    688         Some(OperationalListingAvailability::Status { status }) => {
    689             matches!(status, OperationalListingStatus::Active)
    690         }
    691         Some(OperationalListingAvailability::Window { .. }) | None => true,
    692     }
    693 }
    694 
    695 fn trade_product_fields_from_listing(
    696     listing: &OperationalListing,
    697     listing_addr: &str,
    698 ) -> Result<ITradeProductFields, RadrootsReplicaEventsError> {
    699     let bin = primary_listing_bin(listing)?;
    700     let quantity_amount = bin.quantity.amount();
    701     let qty_amt = decimal_to_f64(&quantity_amount, "listing primary bin quantity")?;
    702     let qty_amt_exact = quantity_amount.to_string();
    703     let qty_avail = listing
    704         .inventory_available
    705         .as_ref()
    706         .map(|amount| decimal_to_i64(amount, "listing inventory"))
    707         .transpose()?;
    708     let price_source = bin
    709         .display_price
    710         .as_ref()
    711         .unwrap_or(bin.price_per_canonical_unit.amount());
    712     let Some(price_amt) = price_source.amount().to_f64_lossy() else {
    713         return Err(RadrootsReplicaEventsError::InvalidData(
    714             "listing price amount out of range".to_string(),
    715         ));
    716     };
    717     let price_amt_exact = price_source.amount().to_string();
    718     let price_currency = price_source.currency().as_str().to_string();
    719     let price_qty_amt = if bin.display_price.is_some() {
    720         1.0
    721     } else {
    722         decimal_to_f64(
    723             &bin.price_per_canonical_unit.quantity().amount(),
    724             "listing price quantity",
    725         )?
    726     };
    727     let price_qty_amt_exact = if bin.display_price.is_some() {
    728         "1".to_string()
    729     } else {
    730         bin.price_per_canonical_unit.quantity().amount().to_string()
    731     };
    732     let price_qty_unit = bin
    733         .display_price_unit
    734         .unwrap_or(bin.price_per_canonical_unit.quantity().unit())
    735         .to_string();
    736 
    737     Ok(ITradeProductFields {
    738         key: listing.product.key.clone(),
    739         category: listing.product.category.clone(),
    740         title: listing.product.title.clone(),
    741         summary: listing.product.summary.clone().unwrap_or_default(),
    742         process: listing.product.process.clone().unwrap_or_default(),
    743         lot: listing.product.lot.clone().unwrap_or_default(),
    744         profile: listing.product.profile.clone().unwrap_or_default(),
    745         year: listing
    746             .product
    747             .year
    748             .as_deref()
    749             .and_then(|value| value.parse::<i64>().ok())
    750             .unwrap_or_default(),
    751         qty_amt,
    752         qty_amt_exact,
    753         qty_unit: bin.quantity.unit().to_string(),
    754         qty_label: bin
    755             .display_label
    756             .clone()
    757             .or_else(|| bin.quantity.label().map(ToOwned::to_owned)),
    758         qty_avail,
    759         price_amt,
    760         price_amt_exact,
    761         price_currency,
    762         price_qty_amt,
    763         price_qty_amt_exact,
    764         price_qty_unit,
    765         listing_addr: Some(listing_addr.to_string()),
    766         primary_bin_id: Some(listing.primary_bin_id.to_string()),
    767         verified_primary_bin_id: Some(listing.primary_bin_id.to_string()),
    768         notes: trade_product_notes_from_listing(listing)?,
    769     })
    770 }
    771 
    772 fn trade_product_notes_from_listing(
    773     listing: &OperationalListing,
    774 ) -> Result<Option<String>, RadrootsReplicaEventsError> {
    775     let Some(discounts) = listing
    776         .discounts
    777         .as_ref()
    778         .filter(|discounts| !discounts.is_empty())
    779     else {
    780         return Ok(None);
    781     };
    782     match serde_json::to_string(&json!({ "listing_discounts": discounts })) {
    783         Ok(notes) => Ok(Some(notes)),
    784         Err(error) => Err(RadrootsReplicaEventsError::InvalidData(format!(
    785             "listing discounts could not be serialized: {error}"
    786         ))),
    787     }
    788 }
    789 
    790 fn primary_listing_bin(
    791     listing: &OperationalListing,
    792 ) -> Result<&OperationalListingBin, RadrootsReplicaEventsError> {
    793     listing
    794         .bins
    795         .iter()
    796         .find(|bin| bin.bin_id == listing.primary_bin_id)
    797         .ok_or_else(|| {
    798             RadrootsReplicaEventsError::InvalidData(
    799                 "listing primary bin missing from bins".to_string(),
    800             )
    801         })
    802 }
    803 
    804 fn decimal_to_i64(value: &Decimal, field: &str) -> Result<i64, RadrootsReplicaEventsError> {
    805     let value = decimal_to_u64(value, field)?;
    806     match i64::try_from(value) {
    807         Ok(value) => Ok(value),
    808         Err(_) => Err(RadrootsReplicaEventsError::InvalidData(format!(
    809             "{field} exceeds i64 range"
    810         ))),
    811     }
    812 }
    813 
    814 fn decimal_to_f64(value: &Decimal, field: &str) -> Result<f64, RadrootsReplicaEventsError> {
    815     match value.to_f64_lossy() {
    816         Some(value) => Ok(value),
    817         None => Err(RadrootsReplicaEventsError::InvalidData(format!(
    818             "{field} exceeds f64 range"
    819         ))),
    820     }
    821 }
    822 
    823 fn decimal_to_u64(value: &Decimal, field: &str) -> Result<u64, RadrootsReplicaEventsError> {
    824     match value.to_u64_exact() {
    825         Some(value) => Ok(value),
    826         None => Err(RadrootsReplicaEventsError::InvalidData(format!(
    827             "{field} must be a whole number"
    828         ))),
    829     }
    830 }
    831 
    832 fn trade_product_listing_addr_filter(listing_addr: &str) -> ITradeProductFieldsFilter {
    833     ITradeProductFieldsFilter {
    834         id: None,
    835         created_at: None,
    836         updated_at: None,
    837         key: None,
    838         category: None,
    839         title: None,
    840         summary: None,
    841         process: None,
    842         lot: None,
    843         profile: None,
    844         year: None,
    845         qty_amt: None,
    846         qty_amt_exact: None,
    847         qty_unit: None,
    848         qty_label: None,
    849         qty_avail: None,
    850         price_amt: None,
    851         price_amt_exact: None,
    852         price_currency: None,
    853         price_qty_amt: None,
    854         price_qty_amt_exact: None,
    855         price_qty_unit: None,
    856         listing_addr: Some(listing_addr.to_string()),
    857         primary_bin_id: None,
    858         verified_primary_bin_id: None,
    859         notes: None,
    860     }
    861 }
    862 
    863 fn upsert_trade_product_for_listing_addr(
    864     exec: &dyn SqlExecutor,
    865     listing_addr: &str,
    866     fields: ITradeProductFields,
    867 ) -> Result<(), RadrootsReplicaEventsError> {
    868     let existing = trade_product::find_many(
    869         exec,
    870         &ITradeProductFindMany {
    871             filter: Some(trade_product_listing_addr_filter(listing_addr)),
    872         },
    873     )?
    874     .results;
    875 
    876     if let Some(row) = existing.first() {
    877         let update = ITradeProductUpdate {
    878             on: TradeProductQueryBindValues::Id { id: row.id.clone() },
    879             fields: trade_product_partial_from_fields(&fields),
    880         };
    881         let _ = trade_product::update(exec, &update)?;
    882         for duplicate in existing.iter().skip(1) {
    883             delete_trade_product_by_id(exec, &duplicate.id)?;
    884         }
    885     } else {
    886         let _ = trade_product::create(exec, &fields)?;
    887     }
    888 
    889     Ok(())
    890 }
    891 
    892 fn delete_trade_products_for_listing_addr(
    893     exec: &dyn SqlExecutor,
    894     listing_addr: &str,
    895 ) -> Result<(), RadrootsReplicaEventsError> {
    896     let existing = trade_product::find_many(
    897         exec,
    898         &ITradeProductFindMany {
    899             filter: Some(trade_product_listing_addr_filter(listing_addr)),
    900         },
    901     )?
    902     .results;
    903 
    904     for row in existing {
    905         delete_trade_product_by_id(exec, &row.id)?;
    906     }
    907 
    908     Ok(())
    909 }
    910 
    911 fn delete_trade_product_by_id(
    912     exec: &dyn SqlExecutor,
    913     id: &str,
    914 ) -> Result<(), RadrootsReplicaEventsError> {
    915     let _ = trade_product::delete(
    916         exec,
    917         &ITradeProductFindOne::On(ITradeProductFindOneArgs {
    918             on: TradeProductQueryBindValues::Id { id: id.to_string() },
    919         }),
    920     )?;
    921     Ok(())
    922 }
    923 
    924 fn trade_product_partial_from_fields(fields: &ITradeProductFields) -> ITradeProductFieldsPartial {
    925     ITradeProductFieldsPartial {
    926         key: Some(Value::from(fields.key.clone())),
    927         category: Some(Value::from(fields.category.clone())),
    928         title: Some(Value::from(fields.title.clone())),
    929         summary: Some(Value::from(fields.summary.clone())),
    930         process: Some(Value::from(fields.process.clone())),
    931         lot: Some(Value::from(fields.lot.clone())),
    932         profile: Some(Value::from(fields.profile.clone())),
    933         year: Some(Value::from(fields.year)),
    934         qty_amt: Some(Value::from(fields.qty_amt)),
    935         qty_amt_exact: Some(Value::from(fields.qty_amt_exact.clone())),
    936         qty_unit: Some(Value::from(fields.qty_unit.clone())),
    937         qty_label: to_value_opt(fields.qty_label.clone()),
    938         qty_avail: fields.qty_avail.map(Value::from).or(Some(Value::Null)),
    939         price_amt: Some(Value::from(fields.price_amt)),
    940         price_amt_exact: Some(Value::from(fields.price_amt_exact.clone())),
    941         price_currency: Some(Value::from(fields.price_currency.clone())),
    942         price_qty_amt: Some(Value::from(fields.price_qty_amt)),
    943         price_qty_amt_exact: Some(Value::from(fields.price_qty_amt_exact.clone())),
    944         price_qty_unit: Some(Value::from(fields.price_qty_unit.clone())),
    945         listing_addr: to_value_opt(fields.listing_addr.clone()),
    946         primary_bin_id: to_value_opt(fields.primary_bin_id.clone()),
    947         verified_primary_bin_id: to_value_opt(fields.verified_primary_bin_id.clone()),
    948         notes: to_value_opt(fields.notes.clone()),
    949     }
    950 }
    951 
    952 /// Advances a supported non-classified event head without running projection.
    953 ///
    954 /// Kind 30402 is rejected because its raw head and profile-aware projection
    955 /// cleanup must be applied atomically through full replica ingestion.
    956 pub fn radroots_replica_ingest_event_head(
    957     exec: &dyn SqlExecutor,
    958     event: &EventEnvelope,
    959 ) -> Result<RadrootsReplicaIngestOutcome, RadrootsReplicaEventsError> {
    960     if event.kind_u32() == KIND_CLASSIFIED_LISTING {
    961         verify_nip01_event(event.clone())?;
    962         return Err(RadrootsReplicaEventsError::InvalidData(
    963             "classified listing heads require profile-aware replica ingestion".to_string(),
    964         ));
    965     }
    966     let decision = event_head_decision(exec, event)?;
    967     if !decision.apply {
    968         return Ok(RadrootsReplicaIngestOutcome::Skipped);
    969     }
    970     upsert_event_head(exec, &decision)?;
    971     Ok(RadrootsReplicaIngestOutcome::Applied)
    972 }
    973 
    974 fn upsert_event_head(
    975     exec: &dyn SqlExecutor,
    976     decision: &EventHeadDecision,
    977 ) -> Result<(), RadrootsReplicaEventsError> {
    978     let existing_result = nostr_event_head::find_one(
    979         exec,
    980         &INostrEventHeadFindOne::On(INostrEventHeadFindOneArgs {
    981             on: NostrEventHeadQueryBindValues::Key {
    982                 key: decision.key.clone(),
    983             },
    984         }),
    985     );
    986     let existing = existing_result?.result;
    987 
    988     match existing {
    989         Some(state) => {
    990             let fields = INostrEventHeadFieldsPartial {
    991                 key: None,
    992                 kind: None,
    993                 pubkey: None,
    994                 d_tag: None,
    995                 last_event_id: Some(Value::from(decision.last_event_id.clone())),
    996                 last_created_at: Some(Value::from(decision.last_created_at)),
    997                 content_hash: Some(Value::from(decision.content_hash.clone())),
    998             };
    999             let update_result = nostr_event_head::update(
   1000                 exec,
   1001                 &INostrEventHeadUpdate {
   1002                     on: NostrEventHeadQueryBindValues::Id { id: state.id },
   1003                     fields,
   1004                 },
   1005             );
   1006             let _updated = update_result?;
   1007         }
   1008         None => {
   1009             let fields = INostrEventHeadFields {
   1010                 key: decision.key.clone(),
   1011                 kind: decision.kind,
   1012                 pubkey: decision.pubkey.clone(),
   1013                 d_tag: decision.d_tag.clone(),
   1014                 last_event_id: decision.last_event_id.clone(),
   1015                 last_created_at: decision.last_created_at,
   1016                 content_hash: decision.content_hash.clone(),
   1017             };
   1018             let _ = nostr_event_head::create(exec, &fields)?;
   1019         }
   1020     }
   1021 
   1022     Ok(())
   1023 }
   1024 
   1025 fn event_head_decision(
   1026     exec: &dyn SqlExecutor,
   1027     event: &EventEnvelope,
   1028 ) -> Result<EventHeadDecision, RadrootsReplicaEventsError> {
   1029     let candidate_result = if event.kind_u32() == KIND_CLASSIFIED_LISTING {
   1030         event_head_candidate_for_class(event, EventClass::Addressable)
   1031     } else {
   1032         match event_head_candidate_for_event(event) {
   1033             Ok(candidate) => candidate,
   1034             Err(err) => {
   1035                 return Err(RadrootsReplicaEventsError::InvalidData(format!(
   1036                     "event head contract mismatch: {err:?}"
   1037                 )));
   1038             }
   1039         }
   1040     };
   1041     let candidate = match candidate_result {
   1042         EventHeadCandidateResult::Candidate(candidate) => candidate,
   1043         EventHeadCandidateResult::NotHeadSelected => {
   1044             return Err(RadrootsReplicaEventsError::InvalidData(
   1045                 "event is not head-selected".to_string(),
   1046             ));
   1047         }
   1048         EventHeadCandidateResult::NotPersisted => {
   1049             return Ok(EventHeadDecision {
   1050                 apply: false,
   1051                 key: String::new(),
   1052                 kind: event.kind_u32(),
   1053                 pubkey: event.author().to_hex().to_owned(),
   1054                 d_tag: String::new(),
   1055                 last_event_id: event.id_hex(),
   1056                 last_created_at: event.created_at_u64(),
   1057                 content_hash: String::new(),
   1058             });
   1059         }
   1060         EventHeadCandidateResult::Malformed(err) => {
   1061             return Err(RadrootsReplicaEventsError::InvalidData(format!(
   1062                 "malformed event head: {err:?}"
   1063             )));
   1064         }
   1065     };
   1066     let (key, kind, pubkey, d_tag) = event_head_coordinate_fields(&candidate.coordinate);
   1067     #[cfg(test)]
   1068     let content_hash = event_content_hash(event.content(), &event.tags_as_vec())?;
   1069     #[cfg(not(test))]
   1070     let content_hash = event_content_hash(event.content(), &event.tags_as_vec());
   1071     let existing_result = nostr_event_head::find_one(
   1072         exec,
   1073         &INostrEventHeadFindOne::On(INostrEventHeadFindOneArgs {
   1074             on: NostrEventHeadQueryBindValues::Key { key: key.clone() },
   1075         }),
   1076     );
   1077     let existing = existing_result?.result;
   1078     let current = existing
   1079         .as_ref()
   1080         .map(|state| current_event_head_from_row(state, &candidate.coordinate))
   1081         .transpose()?;
   1082 
   1083     let decision = select_event_head(candidate, current.as_ref());
   1084     let apply = match decision {
   1085         ProtocolEventHeadDecision::Applied(_) => true,
   1086         ProtocolEventHeadDecision::SkippedDuplicate
   1087         | ProtocolEventHeadDecision::SkippedOlder
   1088         | ProtocolEventHeadDecision::SkippedSameTimestampHigherEventId => false,
   1089         ProtocolEventHeadDecision::CoordinateMismatch => {
   1090             return Err(RadrootsReplicaEventsError::InvalidData(
   1091                 "event head coordinate mismatch".to_string(),
   1092             ));
   1093         }
   1094     };
   1095 
   1096     Ok(EventHeadDecision {
   1097         apply,
   1098         key,
   1099         kind,
   1100         pubkey,
   1101         d_tag,
   1102         last_event_id: event.id_hex(),
   1103         last_created_at: event.created_at_u64(),
   1104         content_hash,
   1105     })
   1106 }
   1107 
   1108 fn current_event_head_from_row(
   1109     row: &NostrEventHead,
   1110     coordinate: &EventHeadCoordinate,
   1111 ) -> Result<CurrentEventHead, RadrootsReplicaEventsError> {
   1112     let event_id = EventId::parse(&row.last_event_id).map_err(|err| {
   1113         RadrootsReplicaEventsError::InvalidData(format!(
   1114             "nostr event head last_event_id invalid: {err}"
   1115         ))
   1116     })?;
   1117     Ok(CurrentEventHead {
   1118         coordinate: coordinate.clone(),
   1119         event_id,
   1120         created_at: row.last_created_at,
   1121     })
   1122 }
   1123 
   1124 fn event_head_coordinate_fields(coordinate: &EventHeadCoordinate) -> (String, u32, String, String) {
   1125     match coordinate {
   1126         EventHeadCoordinate::Replaceable { kind, pubkey } => {
   1127             let pubkey = pubkey.to_string();
   1128             (
   1129                 event_head_key(*kind, &pubkey, ""),
   1130                 *kind,
   1131                 pubkey,
   1132                 String::new(),
   1133             )
   1134         }
   1135         EventHeadCoordinate::Addressable {
   1136             kind,
   1137             pubkey,
   1138             d_tag,
   1139         } => {
   1140             let pubkey = pubkey.to_string();
   1141             let d_tag = d_tag.to_string();
   1142             (event_head_key(*kind, &pubkey, &d_tag), *kind, pubkey, d_tag)
   1143         }
   1144     }
   1145 }
   1146 
   1147 fn find_farm_by_ref(
   1148     exec: &dyn SqlExecutor,
   1149     pubkey: &str,
   1150     d_tag: &str,
   1151 ) -> Result<radroots_replica_schema::farm::Farm, RadrootsReplicaEventsError> {
   1152     let filter = IFarmFieldsFilter {
   1153         id: None,
   1154         created_at: None,
   1155         updated_at: None,
   1156         d_tag: Some(d_tag.to_string()),
   1157         pubkey: Some(pubkey.to_string()),
   1158         name: None,
   1159         about: None,
   1160         website: None,
   1161         picture: None,
   1162         banner: None,
   1163         location_primary: None,
   1164         location_city: None,
   1165         location_region: None,
   1166         location_country: None,
   1167     };
   1168     let result_query = farm::find_many(
   1169         exec,
   1170         &IFarmFindMany {
   1171             filter: Some(filter),
   1172         },
   1173     );
   1174     let result = result_query?;
   1175     match result.results.into_iter().next() {
   1176         Some(farm) => Ok(farm),
   1177         None => Err(RadrootsReplicaEventsError::InvalidData(
   1178             "farm not found".to_string(),
   1179         )),
   1180     }
   1181 }
   1182 
   1183 fn upsert_farm_tags(
   1184     exec: &dyn SqlExecutor,
   1185     farm_id: &str,
   1186     tags: Option<Vec<String>>,
   1187 ) -> Result<(), RadrootsReplicaEventsError> {
   1188     let existing_query = farm_tag::find_many(
   1189         exec,
   1190         &IFarmTagFindMany {
   1191             filter: Some(IFarmTagFieldsFilter {
   1192                 id: None,
   1193                 created_at: None,
   1194                 updated_at: None,
   1195                 farm_id: Some(farm_id.to_string()),
   1196                 tag: None,
   1197             }),
   1198         },
   1199     );
   1200     let existing = existing_query?;
   1201     for row in existing.results {
   1202         handle_delete_result(farm_tag::delete(
   1203             exec,
   1204             &IFarmTagDelete::On(IFarmTagFindOneArgs {
   1205                 on: FarmTagQueryBindValues::Id { id: row.id },
   1206             }),
   1207         ))?;
   1208     }
   1209 
   1210     let mut tags = tags.unwrap_or_default();
   1211     tags.sort();
   1212     tags.dedup();
   1213     for tag in tags {
   1214         if tag.trim().is_empty() {
   1215             continue;
   1216         }
   1217         let fields = IFarmTagFields {
   1218             farm_id: farm_id.to_string(),
   1219             tag,
   1220         };
   1221         let _ = farm_tag::create(exec, &fields)?;
   1222     }
   1223     Ok(())
   1224 }
   1225 
   1226 fn upsert_plot_tags(
   1227     exec: &dyn SqlExecutor,
   1228     plot_id: &str,
   1229     tags: Option<Vec<String>>,
   1230 ) -> Result<(), RadrootsReplicaEventsError> {
   1231     let existing_query = plot_tag::find_many(
   1232         exec,
   1233         &IPlotTagFindMany {
   1234             filter: Some(IPlotTagFieldsFilter {
   1235                 id: None,
   1236                 created_at: None,
   1237                 updated_at: None,
   1238                 plot_id: Some(plot_id.to_string()),
   1239                 tag: None,
   1240             }),
   1241         },
   1242     );
   1243     let existing = existing_query?;
   1244     for row in existing.results {
   1245         handle_delete_result(plot_tag::delete(
   1246             exec,
   1247             &IPlotTagDelete::On(IPlotTagFindOneArgs {
   1248                 on: PlotTagQueryBindValues::Id { id: row.id },
   1249             }),
   1250         ))?;
   1251     }
   1252 
   1253     let mut tags = tags.unwrap_or_default();
   1254     tags.sort();
   1255     tags.dedup();
   1256     for tag in tags {
   1257         if tag.trim().is_empty() {
   1258             continue;
   1259         }
   1260         let fields = IPlotTagFields {
   1261             plot_id: plot_id.to_string(),
   1262             tag,
   1263         };
   1264         let _ = plot_tag::create(exec, &fields)?;
   1265     }
   1266     Ok(())
   1267 }
   1268 
   1269 fn upsert_plot_location(
   1270     exec: &dyn SqlExecutor,
   1271     plot_id: &str,
   1272     location: Option<radroots_event::farm::plot::PlotLocation>,
   1273     factory: &dyn RadrootsReplicaIdFactory,
   1274 ) -> Result<(), RadrootsReplicaEventsError> {
   1275     clear_plot_locations(exec, plot_id)?;
   1276     if let Some(location) = location {
   1277         let gcs_id = create_gcs_location(exec, location.gcs, factory)?;
   1278         let fields = IPlotGcsLocationFields {
   1279             plot_id: plot_id.to_string(),
   1280             gcs_location_id: gcs_id,
   1281             role: ROLE_PRIMARY.to_string(),
   1282         };
   1283         let _ = plot_gcs_location::create(exec, &fields)?;
   1284     }
   1285     Ok(())
   1286 }
   1287 
   1288 #[cfg(test)]
   1289 fn clear_farm_locations(
   1290     exec: &dyn SqlExecutor,
   1291     farm_id: &str,
   1292 ) -> Result<(), RadrootsReplicaEventsError> {
   1293     let existing_query = farm_gcs_location::find_many(
   1294         exec,
   1295         &IFarmGcsLocationFindMany {
   1296             filter: Some(IFarmGcsLocationFieldsFilter {
   1297                 id: None,
   1298                 created_at: None,
   1299                 updated_at: None,
   1300                 farm_id: Some(farm_id.to_string()),
   1301                 gcs_location_id: None,
   1302                 role: None,
   1303             }),
   1304         },
   1305     );
   1306     let existing = existing_query?;
   1307     for row in existing.results {
   1308         handle_delete_result(farm_gcs_location::delete(
   1309             exec,
   1310             &IFarmGcsLocationDelete::On(IFarmGcsLocationFindOneArgs {
   1311                 on: FarmGcsLocationQueryBindValues::Id { id: row.id },
   1312             }),
   1313         ))?;
   1314     }
   1315     Ok(())
   1316 }
   1317 
   1318 fn clear_plot_locations(
   1319     exec: &dyn SqlExecutor,
   1320     plot_id: &str,
   1321 ) -> Result<(), RadrootsReplicaEventsError> {
   1322     let existing_query = plot_gcs_location::find_many(
   1323         exec,
   1324         &IPlotGcsLocationFindMany {
   1325             filter: Some(IPlotGcsLocationFieldsFilter {
   1326                 id: None,
   1327                 created_at: None,
   1328                 updated_at: None,
   1329                 plot_id: Some(plot_id.to_string()),
   1330                 gcs_location_id: None,
   1331                 role: None,
   1332             }),
   1333         },
   1334     );
   1335     let existing = existing_query?;
   1336     for row in existing.results {
   1337         handle_delete_result(plot_gcs_location::delete(
   1338             exec,
   1339             &IPlotGcsLocationDelete::On(IPlotGcsLocationFindOneArgs {
   1340                 on: PlotGcsLocationQueryBindValues::Id { id: row.id },
   1341             }),
   1342         ))?;
   1343     }
   1344     Ok(())
   1345 }
   1346 
   1347 fn create_gcs_location(
   1348     exec: &dyn SqlExecutor,
   1349     gcs: radroots_event::farm::change_set::GcsLocation,
   1350     factory: &dyn RadrootsReplicaIdFactory,
   1351 ) -> Result<String, RadrootsReplicaEventsError> {
   1352     let d_tag = factory.new_d_tag();
   1353     #[cfg(test)]
   1354     let point = serialize_gcs_point(&gcs.point).map_err(map_gcs_point_serialize_error)?;
   1355     #[cfg(not(test))]
   1356     let point = serialize_gcs_point(&gcs.point);
   1357 
   1358     #[cfg(test)]
   1359     let polygon = serialize_gcs_polygon(&gcs.polygon).map_err(map_gcs_polygon_serialize_error)?;
   1360     #[cfg(not(test))]
   1361     let polygon = serialize_gcs_polygon(&gcs.polygon);
   1362 
   1363     let fields = IGcsLocationFields {
   1364         d_tag,
   1365         lat: gcs.lat,
   1366         lng: gcs.lng,
   1367         geohash: gcs.geohash,
   1368         point,
   1369         polygon,
   1370         accuracy: gcs.accuracy,
   1371         altitude: gcs.altitude,
   1372         tag_0: gcs.tag_0,
   1373         label: gcs.label,
   1374         area: gcs.area,
   1375         elevation: gcs.elevation,
   1376         soil: gcs.soil,
   1377         climate: gcs.climate,
   1378         gc_id: gcs.gc_id,
   1379         gc_name: gcs.gc_name,
   1380         gc_admin1_id: gcs.gc_admin1_id,
   1381         gc_admin1_name: gcs.gc_admin1_name,
   1382         gc_country_id: gcs.gc_country_id,
   1383         gc_country_name: gcs.gc_country_name,
   1384     };
   1385     let result = gcs_location::create(exec, &fields)?;
   1386     Ok(result.result.id)
   1387 }
   1388 
   1389 #[cfg(test)]
   1390 fn map_gcs_point_serialize_error(_err: serde_json::Error) -> RadrootsReplicaEventsError {
   1391     RadrootsReplicaEventsError::InvalidData("gcs.point".to_string())
   1392 }
   1393 
   1394 #[cfg(test)]
   1395 fn map_gcs_polygon_serialize_error(_err: serde_json::Error) -> RadrootsReplicaEventsError {
   1396     RadrootsReplicaEventsError::InvalidData("gcs.polygon".to_string())
   1397 }
   1398 
   1399 #[cfg(test)]
   1400 fn serialize_gcs_point(
   1401     point: &radroots_event::farm::change_set::GeoJsonPoint,
   1402 ) -> Result<String, serde_json::Error> {
   1403     #[cfg(test)]
   1404     if failpoints::take_gcs_point_serialize_error() {
   1405         return Err(json_parse_error());
   1406     }
   1407     serde_json::to_string(point)
   1408 }
   1409 
   1410 #[cfg(not(test))]
   1411 fn serialize_gcs_point(point: &radroots_event::farm::change_set::GeoJsonPoint) -> String {
   1412     serde_json::to_string(point).expect("gcs.point serializes")
   1413 }
   1414 
   1415 #[cfg(test)]
   1416 fn serialize_gcs_polygon(
   1417     polygon: &radroots_event::farm::change_set::GeoJsonPolygon,
   1418 ) -> Result<String, serde_json::Error> {
   1419     #[cfg(test)]
   1420     if failpoints::take_gcs_polygon_serialize_error() {
   1421         return Err(json_parse_error());
   1422     }
   1423     serde_json::to_string(polygon)
   1424 }
   1425 
   1426 #[cfg(not(test))]
   1427 fn serialize_gcs_polygon(polygon: &radroots_event::farm::change_set::GeoJsonPolygon) -> String {
   1428     serde_json::to_string(polygon).expect("gcs.polygon serializes")
   1429 }
   1430 
   1431 #[cfg(test)]
   1432 fn json_parse_error() -> serde_json::Error {
   1433     serde_json::from_str::<Value>("{").expect_err("json parse error")
   1434 }
   1435 
   1436 fn upsert_farm_members(
   1437     exec: &dyn SqlExecutor,
   1438     farm_id: &str,
   1439     role: ListSetRole,
   1440     list_set: &radroots_event::social::list_set::ListSet,
   1441 ) -> Result<(), RadrootsReplicaEventsError> {
   1442     let role_value = match role {
   1443         ListSetRole::Members => ROLE_MEMBER,
   1444         ListSetRole::Owners => ROLE_OWNER,
   1445         ListSetRole::Workers => ROLE_WORKER,
   1446         ListSetRole::Plots => return Ok(()),
   1447     };
   1448     let existing_query = farm_member::find_many(
   1449         exec,
   1450         &IFarmMemberFindMany {
   1451             filter: Some(IFarmMemberFieldsFilter {
   1452                 id: None,
   1453                 created_at: None,
   1454                 updated_at: None,
   1455                 farm_id: Some(farm_id.to_string()),
   1456                 member_pubkey: None,
   1457                 role: Some(role_value.to_string()),
   1458             }),
   1459         },
   1460     );
   1461     let existing = existing_query?;
   1462     for row in existing.results {
   1463         handle_delete_result(farm_member::delete(
   1464             exec,
   1465             &IFarmMemberDelete::On(IFarmMemberFindOneArgs {
   1466                 on: FarmMemberQueryBindValues::Id { id: row.id },
   1467             }),
   1468         ))?;
   1469     }
   1470 
   1471     let mut entries = Vec::new();
   1472     for entry in &list_set.entries {
   1473         for value in entry.values.iter().take(1) {
   1474             entries.push(value.to_string());
   1475         }
   1476     }
   1477     entries.sort();
   1478     entries.dedup();
   1479 
   1480     for pubkey in entries {
   1481         let fields = IFarmMemberFields {
   1482             farm_id: farm_id.to_string(),
   1483             member_pubkey: pubkey,
   1484             role: role_value.to_string(),
   1485         };
   1486         let _ = farm_member::create(exec, &fields)?;
   1487     }
   1488     Ok(())
   1489 }
   1490 
   1491 fn upsert_member_claims(
   1492     exec: &dyn SqlExecutor,
   1493     member_pubkey: &str,
   1494     list_set: &radroots_event::social::list_set::ListSet,
   1495 ) -> Result<(), RadrootsReplicaEventsError> {
   1496     let existing_query = farm_member_claim::find_many(
   1497         exec,
   1498         &IFarmMemberClaimFindMany {
   1499             filter: Some(IFarmMemberClaimFieldsFilter {
   1500                 id: None,
   1501                 created_at: None,
   1502                 updated_at: None,
   1503                 member_pubkey: Some(member_pubkey.to_string()),
   1504                 farm_pubkey: None,
   1505             }),
   1506         },
   1507     );
   1508     let existing = existing_query?;
   1509     for row in existing.results {
   1510         handle_delete_result(farm_member_claim::delete(
   1511             exec,
   1512             &IFarmMemberClaimDelete::On(IFarmMemberClaimFindOneArgs {
   1513                 on: FarmMemberClaimQueryBindValues::Id { id: row.id },
   1514             }),
   1515         ))?;
   1516     }
   1517 
   1518     let mut entries = Vec::new();
   1519     for entry in &list_set.entries {
   1520         for value in entry.values.iter().take(1) {
   1521             entries.push(value.to_string());
   1522         }
   1523     }
   1524     entries.sort();
   1525     entries.dedup();
   1526 
   1527     for farm_pubkey in entries {
   1528         let fields = IFarmMemberClaimFields {
   1529             member_pubkey: member_pubkey.to_string(),
   1530             farm_pubkey,
   1531         };
   1532         let _ = farm_member_claim::create(exec, &fields)?;
   1533     }
   1534     Ok(())
   1535 }
   1536 
   1537 fn handle_delete_result<T>(
   1538     result: Result<T, ReplicaSchemaError<SqlError>>,
   1539 ) -> Result<(), RadrootsReplicaEventsError> {
   1540     match result {
   1541         Ok(_) => Ok(()),
   1542         Err(err) => {
   1543             if matches!(err.error, SqlError::NotFound(_)) {
   1544                 return Ok(());
   1545             }
   1546             Err(err.into())
   1547         }
   1548     }
   1549 }
   1550 
   1551 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
   1552 enum ListSetRole {
   1553     Members,
   1554     Owners,
   1555     Workers,
   1556     Plots,
   1557 }
   1558 
   1559 fn unpack_farm_location_strings(
   1560     location: Option<&radroots_event::farm::FarmPublicLocation>,
   1561 ) -> (
   1562     Option<String>,
   1563     Option<String>,
   1564     Option<String>,
   1565     Option<String>,
   1566 ) {
   1567     match location {
   1568         Some(location) => (
   1569             Some(location.primary.clone()),
   1570             location.city.clone(),
   1571             location.region.clone(),
   1572             location.country.clone(),
   1573         ),
   1574         None => (None, None, None, None),
   1575     }
   1576 }
   1577 
   1578 fn unpack_plot_location_strings(
   1579     location: Option<&radroots_event::farm::plot::PlotLocation>,
   1580 ) -> (
   1581     Option<String>,
   1582     Option<String>,
   1583     Option<String>,
   1584     Option<String>,
   1585 ) {
   1586     match location {
   1587         Some(location) => (
   1588             location.primary.clone(),
   1589             location.city.clone(),
   1590             location.region.clone(),
   1591             location.country.clone(),
   1592         ),
   1593         None => (None, None, None, None),
   1594     }
   1595 }
   1596 
   1597 fn ensure_list_set_entries_tag(
   1598     list_set: &radroots_event::social::list_set::ListSet,
   1599     expected: &str,
   1600     label: &str,
   1601 ) -> Result<(), RadrootsReplicaEventsError> {
   1602     for entry in list_set.entries.iter() {
   1603         if entry.tag != expected {
   1604             return Err(RadrootsReplicaEventsError::InvalidData(format!(
   1605                 "domain:farm list set {label} must only include {expected} tags"
   1606             )));
   1607         }
   1608     }
   1609     Ok(())
   1610 }
   1611 
   1612 fn parse_farm_list_set_d_tag(d_tag: &str) -> Option<(String, ListSetRole)> {
   1613     let mut parts = d_tag.splitn(3, ':');
   1614     if parts.next() != Some("farm") {
   1615         return None;
   1616     }
   1617     let farm_d_tag = parts.next()?.to_string();
   1618     let suffix = parts.next()?;
   1619     let role = match suffix {
   1620         "members" => ListSetRole::Members,
   1621         "members.owners" => ListSetRole::Owners,
   1622         "members.workers" => ListSetRole::Workers,
   1623         "plots" => ListSetRole::Plots,
   1624         _ => return None,
   1625     };
   1626     Some((farm_d_tag, role))
   1627 }
   1628 
   1629 fn to_value_opt(value: Option<String>) -> Option<Value> {
   1630     Some(match value {
   1631         Some(value) => Value::from(value),
   1632         None => Value::Null,
   1633     })
   1634 }
   1635 
   1636 struct EventHeadDecision {
   1637     apply: bool,
   1638     key: String,
   1639     kind: u32,
   1640     pubkey: String,
   1641     d_tag: String,
   1642     last_event_id: String,
   1643     last_created_at: u64,
   1644     content_hash: String,
   1645 }
   1646 
   1647 #[cfg(test)]
   1648 mod tests {
   1649     use super::*;
   1650     use std::sync::Arc;
   1651     use std::sync::atomic::{AtomicUsize, Ordering};
   1652 
   1653     use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp};
   1654     use radroots_core::{Currency, Money, Quantity, QuantityPrice, Unit};
   1655     use radroots_event::envelope::kind::{KIND_LIST_SET_FOLLOW, KIND_LIST_SET_GENERIC};
   1656     use radroots_event::farm::change_set::{GcsLocation, GeoJsonPoint, GeoJsonPolygon};
   1657     use radroots_event::farm::plot::{Plot, PlotLocation};
   1658     use radroots_event::farm::{Farm, FarmPublicLocation, FarmRef};
   1659     use radroots_event::listing::operational::OperationalListingProduct;
   1660     use radroots_event::profile::{
   1661         ProfileType, RADROOTS_PROFILE_TYPE_TAG_KEY, radroots_profile_type_tag_value,
   1662     };
   1663     use radroots_event::social::list::ListEntry;
   1664     use radroots_event::social::list_set::ListSet;
   1665     use radroots_event_codec::encode::farm as farm_encode;
   1666     use radroots_event_codec::encode::farm as farm_list_sets;
   1667     use radroots_event_codec::encode::list_set as list_set_encode;
   1668     use radroots_event_codec::encode::plot as plot_encode;
   1669     use radroots_nostr::event::from_nostr;
   1670     use radroots_replica_schema::farm::IFarmFields;
   1671     use radroots_replica_schema::farm_gcs_location::IFarmGcsLocationFields;
   1672     use radroots_replica_schema::farm_member::IFarmMemberFields;
   1673     use radroots_replica_schema::farm_member_claim::IFarmMemberClaimFields;
   1674     use radroots_replica_schema::farm_tag::IFarmTagFields;
   1675     use radroots_replica_schema::gcs_location::IGcsLocationFields;
   1676     use radroots_replica_schema::plot::IPlotFields;
   1677     use radroots_replica_schema::plot_gcs_location::IPlotGcsLocationFields;
   1678     use radroots_replica_schema::plot_tag::IPlotTagFields;
   1679     use radroots_replica_store::{
   1680         ReplicaSql, farm, farm_gcs_location, farm_member, farm_member_claim, farm_tag,
   1681         gcs_location, migrations, nostr_event_head, plot, plot_gcs_location, plot_tag,
   1682         trade_product,
   1683     };
   1684     use radroots_sql_core::{ExecOutcome, SqlExecutor, SqlxSqliteExecutor};
   1685     use radroots_test_fixtures::{FIXTURE_ALICE_PUBLIC_KEY_HEX, FIXTURE_ALICE_SECRET_KEY_HEX};
   1686 
   1687     fn fixture_public_key(seed: u8) -> String {
   1688         Keys::parse(&format!("{seed:064x}"))
   1689             .expect("fixture secret key")
   1690             .public_key()
   1691             .to_hex()
   1692     }
   1693 
   1694     fn test_event_envelope(
   1695         id: u64,
   1696         author: &str,
   1697         created_at: u64,
   1698         kind: u32,
   1699         tags: Vec<Vec<String>>,
   1700         content: String,
   1701     ) -> EventEnvelope {
   1702         EventEnvelope::new(EventEnvelopeParts {
   1703             id: format!("{id:064x}"),
   1704             author: author.to_string(),
   1705             created_at,
   1706             kind,
   1707             tags,
   1708             content,
   1709             sig: "f".repeat(128),
   1710         })
   1711         .expect("test event envelope")
   1712     }
   1713 
   1714     fn test_event_with_parts(
   1715         event: &EventEnvelope,
   1716         kind: u32,
   1717         tags: Vec<Vec<String>>,
   1718         content: String,
   1719     ) -> EventEnvelope {
   1720         EventEnvelope::new(EventEnvelopeParts {
   1721             id: event.id_hex(),
   1722             author: event.author().to_hex().to_owned(),
   1723             created_at: event.created_at_u64(),
   1724             kind,
   1725             tags,
   1726             content,
   1727             sig: event.signature_hex(),
   1728         })
   1729         .expect("test event envelope parts")
   1730     }
   1731 
   1732     fn test_event_with_content(event: &EventEnvelope, content: String) -> EventEnvelope {
   1733         test_event_with_parts(event, event.kind_u32(), event.tags_as_vec(), content)
   1734     }
   1735 
   1736     fn test_event_with_id(event: &EventEnvelope, id: String) -> EventEnvelope {
   1737         EventEnvelope::new(EventEnvelopeParts {
   1738             id,
   1739             author: event.author().to_hex().to_owned(),
   1740             created_at: event.created_at_u64(),
   1741             kind: event.kind_u32(),
   1742             tags: event.tags_as_vec(),
   1743             content: event.content().to_owned(),
   1744             sig: event.signature_hex(),
   1745         })
   1746         .expect("test event id")
   1747     }
   1748 
   1749     struct FixedFactory;
   1750 
   1751     impl RadrootsReplicaIdFactory for FixedFactory {
   1752         fn new_d_tag(&self) -> String {
   1753             "AAAAAAAAAAAAAAAAAAAAAZ".to_string()
   1754         }
   1755     }
   1756 
   1757     struct TxnExecutor<'a> {
   1758         inner: Option<&'a SqlxSqliteExecutor>,
   1759         begin_err: Option<SqlError>,
   1760         commit_err: Option<SqlError>,
   1761         rollback_count: Arc<AtomicUsize>,
   1762     }
   1763 
   1764     impl SqlExecutor for TxnExecutor<'_> {
   1765         fn exec(&self, sql: &str, params_json: &str) -> Result<ExecOutcome, SqlError> {
   1766             match self.inner {
   1767                 Some(inner) => inner.exec(sql, params_json),
   1768                 None => Err(SqlError::UnsupportedPlatform),
   1769             }
   1770         }
   1771 
   1772         fn query_raw(&self, sql: &str, params_json: &str) -> Result<String, SqlError> {
   1773             match self.inner {
   1774                 Some(inner) => inner.query_raw(sql, params_json),
   1775                 None => Err(SqlError::UnsupportedPlatform),
   1776             }
   1777         }
   1778 
   1779         fn begin(&self) -> Result<(), SqlError> {
   1780             match self.begin_err.clone() {
   1781                 Some(err) => Err(err),
   1782                 None => Ok(()),
   1783             }
   1784         }
   1785 
   1786         fn commit(&self) -> Result<(), SqlError> {
   1787             match self.commit_err.clone() {
   1788                 Some(err) => Err(err),
   1789                 None => Ok(()),
   1790             }
   1791         }
   1792 
   1793         fn rollback(&self) -> Result<(), SqlError> {
   1794             self.rollback_count.fetch_add(1, Ordering::SeqCst);
   1795             Ok(())
   1796         }
   1797     }
   1798 
   1799     struct DeleteErrorExecutor<'a> {
   1800         inner: &'a SqlxSqliteExecutor,
   1801         table_name: &'static str,
   1802         err: SqlError,
   1803     }
   1804 
   1805     impl SqlExecutor for DeleteErrorExecutor<'_> {
   1806         fn exec(&self, sql: &str, params_json: &str) -> Result<ExecOutcome, SqlError> {
   1807             let normalized = sql.to_ascii_lowercase();
   1808             if normalized.contains("delete from") && normalized.contains(self.table_name) {
   1809                 return Err(self.err.clone());
   1810             }
   1811             self.inner.exec(sql, params_json)
   1812         }
   1813 
   1814         fn query_raw(&self, sql: &str, params_json: &str) -> Result<String, SqlError> {
   1815             self.inner.query_raw(sql, params_json)
   1816         }
   1817 
   1818         fn begin(&self) -> Result<(), SqlError> {
   1819             self.inner.begin()
   1820         }
   1821 
   1822         fn commit(&self) -> Result<(), SqlError> {
   1823             self.inner.commit()
   1824         }
   1825 
   1826         fn rollback(&self) -> Result<(), SqlError> {
   1827             self.inner.rollback()
   1828         }
   1829     }
   1830 
   1831     struct PassExecutor<'a> {
   1832         inner: &'a SqlxSqliteExecutor,
   1833     }
   1834 
   1835     impl SqlExecutor for PassExecutor<'_> {
   1836         fn exec(&self, sql: &str, params_json: &str) -> Result<ExecOutcome, SqlError> {
   1837             self.inner.exec(sql, params_json)
   1838         }
   1839 
   1840         fn query_raw(&self, sql: &str, params_json: &str) -> Result<String, SqlError> {
   1841             self.inner.query_raw(sql, params_json)
   1842         }
   1843 
   1844         fn begin(&self) -> Result<(), SqlError> {
   1845             self.inner.begin()
   1846         }
   1847 
   1848         fn commit(&self) -> Result<(), SqlError> {
   1849             self.inner.commit()
   1850         }
   1851 
   1852         fn rollback(&self) -> Result<(), SqlError> {
   1853             self.inner.rollback()
   1854         }
   1855     }
   1856 
   1857     struct QueryFailExecutor<'a> {
   1858         inner: &'a SqlxSqliteExecutor,
   1859         needle: &'static str,
   1860         err: SqlError,
   1861     }
   1862 
   1863     impl SqlExecutor for QueryFailExecutor<'_> {
   1864         fn exec(&self, sql: &str, params_json: &str) -> Result<ExecOutcome, SqlError> {
   1865             let normalized = sql.to_ascii_lowercase();
   1866             if normalized.contains(self.needle) {
   1867                 return Err(self.err.clone());
   1868             }
   1869             self.inner.exec(sql, params_json)
   1870         }
   1871 
   1872         fn query_raw(&self, sql: &str, params_json: &str) -> Result<String, SqlError> {
   1873             let normalized = sql.to_ascii_lowercase();
   1874             if normalized.contains(self.needle) {
   1875                 return Err(self.err.clone());
   1876             }
   1877             self.inner.query_raw(sql, params_json)
   1878         }
   1879 
   1880         fn begin(&self) -> Result<(), SqlError> {
   1881             self.inner.begin()
   1882         }
   1883 
   1884         fn commit(&self) -> Result<(), SqlError> {
   1885             self.inner.commit()
   1886         }
   1887 
   1888         fn rollback(&self) -> Result<(), SqlError> {
   1889             self.inner.rollback()
   1890         }
   1891     }
   1892 
   1893     fn sample_gcs(lat: f64, lng: f64, geohash: &str) -> GcsLocation {
   1894         GcsLocation {
   1895             lat,
   1896             lng,
   1897             geohash: geohash.to_string(),
   1898             point: GeoJsonPoint {
   1899                 r#type: "Point".to_string(),
   1900                 coordinates: [lng, lat],
   1901             },
   1902             polygon: GeoJsonPolygon {
   1903                 r#type: "Polygon".to_string(),
   1904                 coordinates: vec![vec![
   1905                     [lng, lat],
   1906                     [lng, lat + 0.001],
   1907                     [lng - 0.001, lat + 0.001],
   1908                     [lng, lat],
   1909                 ]],
   1910             },
   1911             accuracy: Some(1.0),
   1912             altitude: Some(2.0),
   1913             tag_0: Some("tag".to_string()),
   1914             label: Some("label".to_string()),
   1915             area: Some(3.0),
   1916             elevation: Some(4),
   1917             soil: Some("soil".to_string()),
   1918             climate: Some("climate".to_string()),
   1919             gc_id: Some("gc_id".to_string()),
   1920             gc_name: Some("gc_name".to_string()),
   1921             gc_admin1_id: Some("gc_admin1_id".to_string()),
   1922             gc_admin1_name: Some("gc_admin1_name".to_string()),
   1923             gc_country_id: Some("gc_country_id".to_string()),
   1924             gc_country_name: Some("gc_country_name".to_string()),
   1925         }
   1926     }
   1927 
   1928     fn profile_event(
   1929         id: u64,
   1930         author: &str,
   1931         created_at: u32,
   1932         profile_type: Option<ProfileType>,
   1933         name: &str,
   1934     ) -> EventEnvelope {
   1935         let profile = serde_json::json!({
   1936             "name": name,
   1937             "display_name": format!("{name}-display"),
   1938             "nip05": format!("{name}@example.com"),
   1939             "about": format!("{name}-about"),
   1940             "website": "https://example.com",
   1941             "picture": "https://example.com/p.png",
   1942             "banner": "https://example.com/b.png",
   1943             "lud06": "lud06",
   1944             "lud16": "lud16"
   1945         });
   1946         let mut tags = Vec::new();
   1947         if let Some(profile_type) = profile_type {
   1948             tags.push(vec![
   1949                 RADROOTS_PROFILE_TYPE_TAG_KEY.to_string(),
   1950                 radroots_profile_type_tag_value(profile_type).to_string(),
   1951             ]);
   1952         }
   1953         test_event_envelope(
   1954             id,
   1955             author,
   1956             u64::from(created_at),
   1957             KIND_PROFILE,
   1958             tags,
   1959             profile.to_string(),
   1960         )
   1961     }
   1962 
   1963     fn test_keys_for_author(author: &str) -> Keys {
   1964         if author == FIXTURE_ALICE_PUBLIC_KEY_HEX {
   1965             return Keys::parse(FIXTURE_ALICE_SECRET_KEY_HEX).expect("fixture signing key");
   1966         }
   1967         (1_u8..=u8::MAX)
   1968             .find_map(|seed| {
   1969                 let keys = Keys::parse(&format!("{seed:064x}")).ok()?;
   1970                 (keys.public_key().to_hex() == author).then_some(keys)
   1971             })
   1972             .expect("test author must resolve to a fixture signing key")
   1973     }
   1974 
   1975     fn sign_test_event(event: &EventEnvelope) -> EventEnvelope {
   1976         let keys = test_keys_for_author(&event.author().to_hex());
   1977         let tags = event
   1978             .tags_as_vec()
   1979             .into_iter()
   1980             .map(|tag| Tag::parse(tag).expect("test event tag"))
   1981             .collect::<Vec<_>>();
   1982         let event = EventBuilder::new(
   1983             Kind::Custom(u16::try_from(event.kind_u32()).expect("test event kind")),
   1984             event.content(),
   1985         )
   1986         .tags(tags)
   1987         .allow_self_tagging()
   1988         .custom_created_at(Timestamp::from_secs(event.created_at_u64()))
   1989         .sign_with_keys(&keys)
   1990         .expect("signed test event");
   1991         from_nostr(&event).expect("test event adapter")
   1992     }
   1993 
   1994     fn ingest_test_profile(
   1995         exec: &dyn SqlExecutor,
   1996         event: &EventEnvelope,
   1997     ) -> Result<RadrootsReplicaIngestOutcome, RadrootsReplicaEventsError> {
   1998         let verified =
   1999             verify_nip01_event(sign_test_event(event)).expect("verified profile fixture");
   2000         ingest_profile_event(exec, &verified)
   2001     }
   2002 
   2003     fn farm_event(
   2004         id: u64,
   2005         author: &str,
   2006         created_at: u32,
   2007         d_tag: &str,
   2008         name: &str,
   2009         location: Option<FarmPublicLocation>,
   2010         tags: Option<Vec<String>>,
   2011     ) -> EventEnvelope {
   2012         let farm = Farm {
   2013             d_tag: d_tag.to_string(),
   2014             name: name.to_string(),
   2015             about: Some("about".to_string()),
   2016             website: Some("https://farm.example.com".to_string()),
   2017             picture: Some("https://farm.example.com/p.png".to_string()),
   2018             banner: Some("https://farm.example.com/b.png".to_string()),
   2019             location,
   2020             tags,
   2021         };
   2022         let tags = farm_encode::farm_build_tags(&farm).expect("farm tags");
   2023         test_event_envelope(
   2024             id,
   2025             author,
   2026             u64::from(created_at),
   2027             KIND_FARM,
   2028             tags,
   2029             serde_json::to_string(&farm).expect("farm json"),
   2030         )
   2031     }
   2032 
   2033     #[allow(clippy::too_many_arguments)]
   2034     fn plot_event(
   2035         id: u64,
   2036         author: &str,
   2037         created_at: u32,
   2038         d_tag: &str,
   2039         farm_ref: FarmRef,
   2040         name: &str,
   2041         location: Option<PlotLocation>,
   2042         tags: Option<Vec<String>>,
   2043     ) -> EventEnvelope {
   2044         let plot = Plot {
   2045             d_tag: d_tag.to_string(),
   2046             farm: farm_ref,
   2047             name: name.to_string(),
   2048             about: Some("plot-about".to_string()),
   2049             location,
   2050             tags,
   2051         };
   2052         let tags = plot_encode::plot_build_tags(&plot).expect("plot tags");
   2053         test_event_envelope(
   2054             id,
   2055             author,
   2056             u64::from(created_at),
   2057             KIND_PLOT,
   2058             tags,
   2059             serde_json::to_string(&plot).expect("plot json"),
   2060         )
   2061     }
   2062 
   2063     fn list_set_event(
   2064         id: u64,
   2065         author: &str,
   2066         created_at: u32,
   2067         kind: u32,
   2068         list_set: &ListSet,
   2069     ) -> EventEnvelope {
   2070         let parts = list_set_encode::to_wire_parts_with_kind(list_set, kind).expect("list set");
   2071         test_event_envelope(
   2072             id,
   2073             author,
   2074             u64::from(created_at),
   2075             kind,
   2076             parts.tags,
   2077             parts.content,
   2078         )
   2079     }
   2080 
   2081     fn listing_event(
   2082         author: &str,
   2083         created_at: u32,
   2084         d_tag: &str,
   2085         status: &str,
   2086         title: &str,
   2087     ) -> EventEnvelope {
   2088         let farm_d_tag = "AAAAAAAAAAAAAAAAAAAAAA";
   2089         signed_listing_event(
   2090             author,
   2091             u64::from(created_at),
   2092             vec![
   2093                 vec!["d".to_string(), d_tag.to_string()],
   2094                 vec![
   2095                     "a".to_string(),
   2096                     format!("{}:{}:{}", KIND_FARM, author, farm_d_tag),
   2097                 ],
   2098                 vec!["p".to_string(), author.to_string()],
   2099                 vec!["key".to_string(), "pasture-eggs".to_string()],
   2100                 vec!["title".to_string(), title.to_string()],
   2101                 vec!["category".to_string(), "eggs".to_string()],
   2102                 vec!["summary".to_string(), "Pasture-raised eggs".to_string()],
   2103                 vec!["process".to_string(), "washed".to_string()],
   2104                 vec!["lot".to_string(), "lot-a".to_string()],
   2105                 vec!["profile".to_string(), "dozen".to_string()],
   2106                 vec!["year".to_string(), "2026".to_string()],
   2107                 vec!["radroots:primary_bin".to_string(), "bin-a".to_string()],
   2108                 vec![
   2109                     "radroots:bin".to_string(),
   2110                     "bin-a".to_string(),
   2111                     "12".to_string(),
   2112                     "each".to_string(),
   2113                     "12".to_string(),
   2114                     "each".to_string(),
   2115                     "dozen".to_string(),
   2116                 ],
   2117                 vec![
   2118                     "radroots:price".to_string(),
   2119                     "bin-a".to_string(),
   2120                     "6".to_string(),
   2121                     "USD".to_string(),
   2122                     "1".to_string(),
   2123                     "each".to_string(),
   2124                     "6".to_string(),
   2125                     "each".to_string(),
   2126                 ],
   2127                 vec!["inventory".to_string(), "5".to_string()],
   2128                 vec!["status".to_string(), status.to_string()],
   2129             ],
   2130             format!("# {title}"),
   2131         )
   2132     }
   2133 
   2134     fn signed_listing_event(
   2135         author: &str,
   2136         created_at: u64,
   2137         tags: Vec<Vec<String>>,
   2138         content: String,
   2139     ) -> EventEnvelope {
   2140         assert_eq!(
   2141             author, FIXTURE_ALICE_PUBLIC_KEY_HEX,
   2142             "listing tests must use the approved fixture signer"
   2143         );
   2144         let keys = Keys::parse(FIXTURE_ALICE_SECRET_KEY_HEX).expect("fixture signing key");
   2145         let tags = tags
   2146             .into_iter()
   2147             .map(|tag| Tag::parse(tag).expect("listing tag"))
   2148             .collect::<Vec<_>>();
   2149         let event = EventBuilder::new(
   2150             Kind::Custom(u16::try_from(KIND_CLASSIFIED_LISTING).expect("classified listing kind")),
   2151             content,
   2152         )
   2153         .tags(tags)
   2154         .allow_self_tagging()
   2155         .custom_created_at(Timestamp::from_secs(created_at))
   2156         .sign_with_keys(&keys)
   2157         .expect("signed listing event");
   2158         from_nostr(&event).expect("listing event adapter")
   2159     }
   2160 
   2161     fn focused_listing_event(author: &str, created_at: u64, d_tag: &str) -> EventEnvelope {
   2162         focused_listing_event_with_content(
   2163             author,
   2164             created_at,
   2165             d_tag,
   2166             "Carrots available this week.",
   2167         )
   2168     }
   2169 
   2170     fn focused_listing_event_with_content(
   2171         author: &str,
   2172         created_at: u64,
   2173         d_tag: &str,
   2174         content: &str,
   2175     ) -> EventEnvelope {
   2176         signed_listing_event(
   2177             author,
   2178             created_at,
   2179             vec![
   2180                 vec!["d".to_string(), d_tag.to_string()],
   2181                 vec!["title".to_string(), "Nantes Carrots".to_string()],
   2182                 vec!["summary".to_string(), "Fresh bunches".to_string()],
   2183                 vec!["published_at".to_string(), "1".to_string()],
   2184                 vec!["location".to_string(), "Central Saanich, BC".to_string()],
   2185                 vec!["price".to_string(), "3".to_string(), "CAD".to_string()],
   2186                 vec!["radroots:price_unit".to_string(), "lb".to_string()],
   2187                 vec!["status".to_string(), "active".to_string()],
   2188             ],
   2189             content.to_string(),
   2190         )
   2191     }
   2192 
   2193     fn generic_listing_event(author: &str, created_at: u64, d_tag: &str) -> EventEnvelope {
   2194         signed_listing_event(
   2195             author,
   2196             created_at,
   2197             vec![
   2198                 vec!["d".to_string(), d_tag.to_string()],
   2199                 vec!["title".to_string(), "Generic listing".to_string()],
   2200                 vec!["price".to_string(), "3".to_string(), "CAD".to_string()],
   2201             ],
   2202             "A standards-compatible marker-free NIP-99 listing.".to_string(),
   2203         )
   2204     }
   2205 
   2206     fn listing_decimal(raw: &str) -> Decimal {
   2207         raw.parse().expect("decimal")
   2208     }
   2209 
   2210     fn listing_currency() -> Currency {
   2211         "USD".parse().expect("currency")
   2212     }
   2213 
   2214     fn listing_model() -> OperationalListing {
   2215         OperationalListing {
   2216             d_tag: "AAAAAAAAAAAAAAAAAAAAAA".parse().expect("d tag"),
   2217             published_at: Some(1),
   2218             farm: FarmRef {
   2219                 pubkey: fixture_public_key(12),
   2220                 d_tag: "AAAAAAAAAAAAAAAAAAAAAZ".to_string(),
   2221             },
   2222             product: OperationalListingProduct {
   2223                 key: "pasture-eggs".to_string(),
   2224                 title: "Pasture Eggs".to_string(),
   2225                 category: "eggs".to_string(),
   2226                 summary: Some("Pasture-raised eggs".to_string()),
   2227                 process: None,
   2228                 lot: None,
   2229                 location: None,
   2230                 profile: None,
   2231                 year: Some("2026".to_string()),
   2232             },
   2233             primary_bin_id: "bin-a".parse().expect("primary bin id"),
   2234             bins: vec![OperationalListingBin {
   2235                 bin_id: "bin-a".parse().expect("bin id"),
   2236                 quantity: Quantity::try_new(listing_decimal("12"), Unit::Each)
   2237                     .unwrap()
   2238                     .with_label("unit label"),
   2239                 price_per_canonical_unit: QuantityPrice::try_new(
   2240                     Money::try_new(listing_decimal("6"), listing_currency()).unwrap(),
   2241                     Quantity::try_new(listing_decimal("1"), Unit::Each).unwrap(),
   2242                 )
   2243                 .unwrap(),
   2244                 display_amount: None,
   2245                 display_unit: None,
   2246                 display_label: None,
   2247                 display_price: None,
   2248                 display_price_unit: None,
   2249             }],
   2250             resource_area: None,
   2251             plot: None,
   2252             discounts: Some(Vec::new()),
   2253             inventory_available: Some(listing_decimal("5")),
   2254             availability: Some(OperationalListingAvailability::Status {
   2255                 status: OperationalListingStatus::Active,
   2256             }),
   2257             delivery_method: None,
   2258             location: None,
   2259             images: None,
   2260         }
   2261     }
   2262 
   2263     #[test]
   2264     fn listing_field_helpers_cover_optional_label_and_error_paths() {
   2265         let listing = listing_model();
   2266         let fields =
   2267             trade_product_fields_from_listing(&listing, "30402:pubkey:listing").expect("fields");
   2268         assert_eq!(fields.qty_label.as_deref(), Some("unit label"));
   2269         assert_eq!(fields.notes, None);
   2270 
   2271         let mut missing_bin = listing.clone();
   2272         missing_bin.primary_bin_id = "bin-missing".parse().expect("missing bin id");
   2273         let err = trade_product_fields_from_listing(&missing_bin, "30402:pubkey:listing")
   2274             .err()
   2275             .expect("missing primary bin should fail");
   2276         assert!(err.to_string().contains("primary bin missing"));
   2277 
   2278         let mut fractional_inventory = listing;
   2279         fractional_inventory.inventory_available = Some(listing_decimal("1.5"));
   2280         let err = trade_product_fields_from_listing(&fractional_inventory, "30402:pubkey:listing")
   2281             .err()
   2282             .expect("fractional inventory should fail");
   2283         assert!(err.to_string().contains("whole number"));
   2284 
   2285         let err = decimal_to_i64(&listing_decimal("9223372036854775808"), "listing inventory")
   2286             .expect_err("i64 overflow should fail");
   2287         assert!(err.to_string().contains("exceeds i64 range"));
   2288     }
   2289 
   2290     #[test]
   2291     fn current_event_head_reports_invalid_stored_event_id() {
   2292         let row = NostrEventHead {
   2293             id: "head-1".to_string(),
   2294             created_at: "2026-01-01T00:00:00Z".to_string(),
   2295             updated_at: "2026-01-01T00:00:00Z".to_string(),
   2296             key: "profile".to_string(),
   2297             kind: KIND_PROFILE,
   2298             pubkey: fixture_public_key(10),
   2299             d_tag: String::new(),
   2300             last_event_id: "not-an-event-id".to_string(),
   2301             last_created_at: 1,
   2302             content_hash: "hash".to_string(),
   2303         };
   2304         let coordinate = EventHeadCoordinate::Replaceable {
   2305             kind: KIND_PROFILE,
   2306             pubkey: fixture_public_key(10).parse().expect("pubkey"),
   2307         };
   2308 
   2309         let err = current_event_head_from_row(&row, &coordinate)
   2310             .expect_err("invalid stored event id should fail");
   2311         assert!(err.to_string().contains("last_event_id invalid"));
   2312     }
   2313 
   2314     fn seed_rows(exec: &SqlxSqliteExecutor) -> (String, String, String, String) {
   2315         migrations::run_all_up(exec).expect("migrations");
   2316         let farm_row = farm::create(
   2317             exec,
   2318             &IFarmFields {
   2319                 d_tag: "AAAAAAAAAAAAAAAAAAAAAA".to_string(),
   2320                 pubkey: fixture_public_key(15),
   2321                 name: "farm".to_string(),
   2322                 about: None,
   2323                 website: None,
   2324                 picture: None,
   2325                 banner: None,
   2326                 location_primary: None,
   2327                 location_city: None,
   2328                 location_region: None,
   2329                 location_country: None,
   2330             },
   2331         )
   2332         .expect("farm")
   2333         .result;
   2334         let plot_row = plot::create(
   2335             exec,
   2336             &IPlotFields {
   2337                 d_tag: "AAAAAAAAAAAAAAAAAAAAAQ".to_string(),
   2338                 farm_id: farm_row.id.clone(),
   2339                 name: "plot".to_string(),
   2340                 about: None,
   2341                 location_primary: None,
   2342                 location_city: None,
   2343                 location_region: None,
   2344                 location_country: None,
   2345             },
   2346         )
   2347         .expect("plot")
   2348         .result;
   2349         let gcs_row = gcs_location::create(
   2350             exec,
   2351             &IGcsLocationFields {
   2352                 d_tag: "AAAAAAAAAAAAAAAAAAAAAw".to_string(),
   2353                 lat: 1.0,
   2354                 lng: 2.0,
   2355                 geohash: "s0".to_string(),
   2356                 point: "{\"type\":\"Point\",\"coordinates\":[2.0,1.0]}".to_string(),
   2357                 polygon:
   2358                     "{\"type\":\"Polygon\",\"coordinates\":[[[2.0,1.0],[2.1,1.1],[1.9,1.1],[2.0,1.0]]]}".to_string(),
   2359                 accuracy: None,
   2360                 altitude: None,
   2361                 tag_0: None,
   2362                 label: None,
   2363                 area: None,
   2364                 elevation: None,
   2365                 soil: None,
   2366                 climate: None,
   2367                 gc_id: None,
   2368                 gc_name: None,
   2369                 gc_admin1_id: None,
   2370                 gc_admin1_name: None,
   2371                 gc_country_id: None,
   2372                 gc_country_name: None,
   2373             },
   2374         )
   2375         .expect("gcs")
   2376         .result;
   2377 
   2378         let _ = farm_tag::create(
   2379             exec,
   2380             &IFarmTagFields {
   2381                 farm_id: farm_row.id.clone(),
   2382                 tag: "alpha".to_string(),
   2383             },
   2384         )
   2385         .expect("farm tag");
   2386         let _ = plot_tag::create(
   2387             exec,
   2388             &IPlotTagFields {
   2389                 plot_id: plot_row.id.clone(),
   2390                 tag: "beta".to_string(),
   2391             },
   2392         )
   2393         .expect("plot tag");
   2394         let _ = farm_gcs_location::create(
   2395             exec,
   2396             &IFarmGcsLocationFields {
   2397                 farm_id: farm_row.id.clone(),
   2398                 gcs_location_id: gcs_row.id.clone(),
   2399                 role: "primary".to_string(),
   2400             },
   2401         )
   2402         .expect("farm gcs");
   2403         let _ = plot_gcs_location::create(
   2404             exec,
   2405             &IPlotGcsLocationFields {
   2406                 plot_id: plot_row.id.clone(),
   2407                 gcs_location_id: gcs_row.id.clone(),
   2408                 role: "primary".to_string(),
   2409             },
   2410         )
   2411         .expect("plot gcs");
   2412         let _ = farm_member::create(
   2413             exec,
   2414             &IFarmMemberFields {
   2415                 farm_id: farm_row.id.clone(),
   2416                 member_pubkey: fixture_public_key(6),
   2417                 role: "member".to_string(),
   2418             },
   2419         )
   2420         .expect("member");
   2421         let _ = farm_member_claim::create(
   2422             exec,
   2423             &IFarmMemberClaimFields {
   2424                 member_pubkey: fixture_public_key(6),
   2425                 farm_pubkey: farm_row.pubkey.clone(),
   2426             },
   2427         )
   2428         .expect("claim");
   2429         (
   2430             farm_row.id,
   2431             farm_row.pubkey,
   2432             farm_row.d_tag,
   2433             plot_row.d_tag.clone(),
   2434         )
   2435     }
   2436 
   2437     #[test]
   2438     fn ingest_transaction_paths_are_covered() {
   2439         let begin_executor = TxnExecutor {
   2440             inner: None,
   2441             begin_err: Some(SqlError::Internal),
   2442             commit_err: None,
   2443             rollback_count: Arc::new(AtomicUsize::new(0)),
   2444         };
   2445         let event = test_event_envelope(
   2446             1,
   2447             &fixture_public_key(10),
   2448             1,
   2449             KIND_LIST_SET_FOLLOW,
   2450             Vec::new(),
   2451             String::new(),
   2452         );
   2453         let begin_err =
   2454             radroots_replica_ingest_event_with_factory(&begin_executor, &event, &FixedFactory)
   2455                 .expect_err("begin");
   2456         assert!(begin_err.to_string().contains("replica_sync.sql"));
   2457         assert!(begin_executor.commit().is_ok());
   2458         assert_eq!(
   2459             begin_executor
   2460                 .exec("select 1", "[]")
   2461                 .expect_err("exec")
   2462                 .code(),
   2463             "ERR_UNSUPPORTED_PLATFORM"
   2464         );
   2465         assert_eq!(
   2466             begin_executor
   2467                 .query_raw("select 1", "[]")
   2468                 .expect_err("query")
   2469                 .code(),
   2470             "ERR_UNSUPPORTED_PLATFORM"
   2471         );
   2472 
   2473         let rollback_count = Arc::new(AtomicUsize::new(0));
   2474         let commit_executor = TxnExecutor {
   2475             inner: None,
   2476             begin_err: None,
   2477             commit_err: Some(SqlError::Internal),
   2478             rollback_count: rollback_count.clone(),
   2479         };
   2480         let commit_err =
   2481             radroots_replica_ingest_event_with_factory(&commit_executor, &event, &FixedFactory)
   2482                 .expect_err("commit");
   2483         assert!(commit_err.to_string().contains("replica_sync.sql"));
   2484         assert_eq!(rollback_count.load(Ordering::SeqCst), 0);
   2485 
   2486         let rollback_executor = TxnExecutor {
   2487             inner: None,
   2488             begin_err: None,
   2489             commit_err: None,
   2490             rollback_count: Arc::new(AtomicUsize::new(0)),
   2491         };
   2492         let unsupported =
   2493             test_event_envelope(2, &fixture_public_key(10), 2, 42, Vec::new(), String::new());
   2494         let err = radroots_replica_ingest_event_with_factory(
   2495             &rollback_executor,
   2496             &unsupported,
   2497             &FixedFactory,
   2498         )
   2499         .expect_err("rollback");
   2500         assert!(err.to_string().contains("unsupported kind"));
   2501         assert_eq!(rollback_executor.rollback_count.load(Ordering::SeqCst), 1);
   2502     }
   2503 
   2504     #[test]
   2505     fn classified_listing_signature_rejection_precedes_transaction_acquisition() {
   2506         let executor = TxnExecutor {
   2507             inner: None,
   2508             begin_err: Some(SqlError::Internal),
   2509             commit_err: None,
   2510             rollback_count: Arc::new(AtomicUsize::new(0)),
   2511         };
   2512         let signed =
   2513             focused_listing_event(FIXTURE_ALICE_PUBLIC_KEY_HEX, 10, "pre-transaction-check");
   2514         let tampered = test_event_with_content(&signed, "tampered".to_string());
   2515 
   2516         assert!(matches!(
   2517             radroots_replica_ingest_event_with_factory(&executor, &tampered, &FixedFactory),
   2518             Err(RadrootsReplicaEventsError::Verification(_))
   2519         ));
   2520         assert_eq!(executor.rollback_count.load(Ordering::SeqCst), 0);
   2521     }
   2522 
   2523     #[test]
   2524     fn calendar_kind_uses_unsupported_transaction_path() {
   2525         let rollback_count = Arc::new(AtomicUsize::new(0));
   2526         let executor = TxnExecutor {
   2527             inner: None,
   2528             begin_err: None,
   2529             commit_err: None,
   2530             rollback_count: rollback_count.clone(),
   2531         };
   2532         let event = test_event_envelope(
   2533             3,
   2534             &fixture_public_key(10),
   2535             3,
   2536             KIND_CALENDAR,
   2537             Vec::new(),
   2538             String::new(),
   2539         );
   2540 
   2541         let err = radroots_replica_ingest_event_with_factory(&executor, &event, &FixedFactory)
   2542             .expect_err("calendar is not supported by the replica projection");
   2543 
   2544         assert!(matches!(
   2545             err,
   2546             RadrootsReplicaEventsError::InvalidData(ref message)
   2547                 if message == "unsupported kind 31924"
   2548         ));
   2549         assert_eq!(rollback_count.load(Ordering::SeqCst), 1);
   2550     }
   2551 
   2552     #[test]
   2553     fn ingest_core_paths_cover_helpers_and_decisions() {
   2554         let exec = SqlxSqliteExecutor::open_memory().expect("db");
   2555         migrations::run_all_up(&exec).expect("migrations");
   2556 
   2557         let factory = RadrootsReplicaDefaultIdFactory;
   2558         assert_eq!(factory.new_d_tag().len(), 22);
   2559 
   2560         let profile_pubkey = fixture_public_key(9);
   2561         let profile = profile_event(
   2562             10,
   2563             &profile_pubkey,
   2564             1,
   2565             Some(ProfileType::Individual),
   2566             "alice",
   2567         );
   2568         let profile_no_type = profile_event(9, &profile_pubkey, 0, None, "alice-none");
   2569         assert!(ingest_test_profile(&exec, &profile_no_type).is_err());
   2570         let signed_profile = sign_test_event(&profile);
   2571         assert_eq!(
   2572             radroots_replica_ingest_event(&exec, &signed_profile).expect("ingest wrapper"),
   2573             RadrootsReplicaIngestOutcome::Applied
   2574         );
   2575         let profile_update = profile_event(
   2576             11,
   2577             &profile_pubkey,
   2578             2,
   2579             Some(ProfileType::Individual),
   2580             "alice-2",
   2581         );
   2582         assert_eq!(
   2583             ingest_test_profile(&exec, &profile_update).expect("profile update"),
   2584             RadrootsReplicaIngestOutcome::Applied
   2585         );
   2586         assert_eq!(
   2587             ingest_test_profile(&exec, &profile_update).expect("profile skip"),
   2588             RadrootsReplicaIngestOutcome::Skipped
   2589         );
   2590         let profile_older = profile_event(
   2591             8,
   2592             &profile_pubkey,
   2593             1,
   2594             Some(ProfileType::Individual),
   2595             "alice-old",
   2596         );
   2597         let decision_old = event_head_decision(&exec, &profile_older).expect("decision old");
   2598         assert!(!decision_old.apply);
   2599         let signed_profile_update = sign_test_event(&profile_update);
   2600         let decision_same =
   2601             event_head_decision(&exec, &signed_profile_update).expect("decision same");
   2602         assert!(!decision_same.apply);
   2603         let profile_same_time_higher_id =
   2604             test_event_with_id(&signed_profile_update, "f".repeat(64));
   2605         let decision = event_head_decision(&exec, &profile_same_time_higher_id).expect("decision");
   2606         assert!(!decision.apply);
   2607         let profile_same_time_lower_id = test_event_with_id(&signed_profile_update, "0".repeat(64));
   2608         let decision = event_head_decision(&exec, &profile_same_time_lower_id).expect("decision");
   2609         assert!(decision.apply);
   2610 
   2611         let farm_pubkey = fixture_public_key(15);
   2612         let farm_d_tag = "AAAAAAAAAAAAAAAAAAAAAA";
   2613         let farm = farm_event(
   2614             20,
   2615             &farm_pubkey,
   2616             10,
   2617             farm_d_tag,
   2618             "farm-a",
   2619             Some(FarmPublicLocation {
   2620                 primary: "primary".to_string(),
   2621                 city: Some("city".to_string()),
   2622                 region: Some("region".to_string()),
   2623                 country: Some("country".to_string()),
   2624                 geohash: "9q8yy".to_string(),
   2625             }),
   2626             Some(vec![
   2627                 "coffee".to_string(),
   2628                 "coffee".to_string(),
   2629                 " ".to_string(),
   2630             ]),
   2631         );
   2632         assert_eq!(
   2633             ingest_farm_event(&exec, &farm, &FixedFactory).expect("farm"),
   2634             RadrootsReplicaIngestOutcome::Applied
   2635         );
   2636         let farm_update = farm_event(
   2637             21,
   2638             &farm_pubkey,
   2639             11,
   2640             farm_d_tag,
   2641             "farm-b",
   2642             None,
   2643             Some(vec!["market".to_string()]),
   2644         );
   2645         assert_eq!(
   2646             ingest_farm_event(&exec, &farm_update, &FixedFactory).expect("farm update"),
   2647             RadrootsReplicaIngestOutcome::Applied
   2648         );
   2649         assert_eq!(
   2650             ingest_farm_event(&exec, &farm_update, &FixedFactory).expect("farm skip"),
   2651             RadrootsReplicaIngestOutcome::Skipped
   2652         );
   2653 
   2654         let plot_d_tag = "AAAAAAAAAAAAAAAAAAAAAQ";
   2655         let plot = plot_event(
   2656             30,
   2657             &farm_pubkey,
   2658             20,
   2659             plot_d_tag,
   2660             FarmRef {
   2661                 pubkey: farm_pubkey.clone(),
   2662                 d_tag: farm_d_tag.to_string(),
   2663             },
   2664             "plot-a",
   2665             Some(PlotLocation {
   2666                 primary: Some("p".to_string()),
   2667                 city: Some("c".to_string()),
   2668                 region: Some("r".to_string()),
   2669                 country: Some("k".to_string()),
   2670                 gcs: sample_gcs(11.0, 21.0, "s1"),
   2671             }),
   2672             Some(vec!["tag".to_string()]),
   2673         );
   2674         assert_eq!(
   2675             ingest_plot_event(&exec, &plot, &FixedFactory).expect("plot"),
   2676             RadrootsReplicaIngestOutcome::Applied
   2677         );
   2678         let plot_update = plot_event(
   2679             31,
   2680             &farm_pubkey,
   2681             21,
   2682             plot_d_tag,
   2683             FarmRef {
   2684                 pubkey: farm_pubkey.clone(),
   2685                 d_tag: farm_d_tag.to_string(),
   2686             },
   2687             "plot-b",
   2688             None,
   2689             Some(vec!["tag2".to_string()]),
   2690         );
   2691         assert_eq!(
   2692             ingest_plot_event(&exec, &plot_update, &FixedFactory).expect("plot update"),
   2693             RadrootsReplicaIngestOutcome::Applied
   2694         );
   2695         assert_eq!(
   2696             ingest_plot_event(&exec, &plot_update, &FixedFactory).expect("plot skip"),
   2697             RadrootsReplicaIngestOutcome::Skipped
   2698         );
   2699 
   2700         let members =
   2701             farm_list_sets::farm_members_list_set(farm_d_tag, vec![fixture_public_key(6)])
   2702                 .expect("members");
   2703         let owners = farm_list_sets::farm_owners_list_set(farm_d_tag, vec![fixture_public_key(8)])
   2704             .expect("owners");
   2705         let workers =
   2706             farm_list_sets::farm_workers_list_set(farm_d_tag, vec![fixture_public_key(16)])
   2707                 .expect("workers");
   2708         let plots = farm_list_sets::farm_plots_list_set(
   2709             farm_d_tag,
   2710             &farm_pubkey,
   2711             vec![plot_d_tag.to_string()],
   2712         )
   2713         .expect("plots");
   2714         let member_of =
   2715             farm_list_sets::member_of_farms_list_set(vec![farm_pubkey.clone()]).expect("member_of");
   2716 
   2717         for (idx, list_set) in [members, owners, workers, plots, member_of]
   2718             .iter()
   2719             .enumerate()
   2720         {
   2721             let event = list_set_event(
   2722                 40 + idx as u64,
   2723                 if list_set.d_tag == "member_of.farms" {
   2724                     &profile_pubkey
   2725                 } else {
   2726                     &farm_pubkey
   2727                 },
   2728                 30 + idx as u32,
   2729                 KIND_LIST_SET_GENERIC,
   2730                 list_set,
   2731             );
   2732             assert_eq!(
   2733                 ingest_list_set_event(&exec, &event).expect("list set"),
   2734                 RadrootsReplicaIngestOutcome::Applied
   2735             );
   2736             assert_eq!(
   2737                 ingest_list_set_event(&exec, &event).expect("list set skip"),
   2738                 RadrootsReplicaIngestOutcome::Skipped
   2739             );
   2740         }
   2741 
   2742         let bad_description = ListSet {
   2743             d_tag: "member_of.farms".to_string(),
   2744             content: String::new(),
   2745             entries: vec![ListEntry {
   2746                 tag: "p".to_string(),
   2747                 values: vec![farm_pubkey.clone()],
   2748             }],
   2749             title: None,
   2750             description: Some("bad".to_string()),
   2751             image: None,
   2752         };
   2753         let bad_description_event = list_set_event(
   2754             90,
   2755             &profile_pubkey,
   2756             100,
   2757             KIND_LIST_SET_GENERIC,
   2758             &bad_description,
   2759         );
   2760         assert!(ingest_list_set_event(&exec, &bad_description_event).is_err());
   2761 
   2762         let bad_image = ListSet {
   2763             d_tag: "member_of.farms".to_string(),
   2764             content: String::new(),
   2765             entries: vec![ListEntry {
   2766                 tag: "p".to_string(),
   2767                 values: vec![farm_pubkey.clone()],
   2768             }],
   2769             title: None,
   2770             description: None,
   2771             image: Some("bad".to_string()),
   2772         };
   2773         let bad_image_event =
   2774             list_set_event(91, &profile_pubkey, 101, KIND_LIST_SET_GENERIC, &bad_image);
   2775         assert!(ingest_list_set_event(&exec, &bad_image_event).is_err());
   2776 
   2777         let bad_title = ListSet {
   2778             d_tag: "member_of.farms".to_string(),
   2779             content: String::new(),
   2780             entries: vec![ListEntry {
   2781                 tag: "p".to_string(),
   2782                 values: vec![farm_pubkey.clone()],
   2783             }],
   2784             title: Some("bad".to_string()),
   2785             description: None,
   2786             image: None,
   2787         };
   2788         let bad_title_event =
   2789             list_set_event(92, &profile_pubkey, 102, KIND_LIST_SET_GENERIC, &bad_title);
   2790         assert!(ingest_list_set_event(&exec, &bad_title_event).is_err());
   2791 
   2792         let bad_content = ListSet {
   2793             d_tag: "member_of.farms".to_string(),
   2794             content: "bad".to_string(),
   2795             entries: vec![ListEntry {
   2796                 tag: "p".to_string(),
   2797                 values: vec![farm_pubkey.clone()],
   2798             }],
   2799             title: None,
   2800             description: None,
   2801             image: None,
   2802         };
   2803         let bad_content_event = list_set_event(
   2804             93,
   2805             &profile_pubkey,
   2806             103,
   2807             KIND_LIST_SET_GENERIC,
   2808             &bad_content,
   2809         );
   2810         assert!(ingest_list_set_event(&exec, &bad_content_event).is_err());
   2811 
   2812         let unknown_farm_list_set = ListSet {
   2813             d_tag: format!("farm:{farm_d_tag}:unknown"),
   2814             content: String::new(),
   2815             entries: vec![ListEntry {
   2816                 tag: "p".to_string(),
   2817                 values: vec![profile_pubkey.clone()],
   2818             }],
   2819             title: None,
   2820             description: None,
   2821             image: None,
   2822         };
   2823         let unknown_farm_list_event = list_set_event(
   2824             94,
   2825             &farm_pubkey,
   2826             104,
   2827             KIND_LIST_SET_GENERIC,
   2828             &unknown_farm_list_set,
   2829         );
   2830         assert!(ingest_list_set_event(&exec, &unknown_farm_list_event).is_err());
   2831 
   2832         assert!(parse_farm_list_set_d_tag("farm:AAAAAAAAAAAAAAAAAAAAAA:unknown").is_none());
   2833         assert!(parse_farm_list_set_d_tag("farm:AAAAAAAAAAAAAAAAAAAAAA:plots").is_some());
   2834         assert_eq!(to_value_opt(Some("x".to_string())), Some(Value::from("x")));
   2835         assert_eq!(to_value_opt(None), Some(Value::Null));
   2836         let location = FarmPublicLocation {
   2837             primary: "p".to_string(),
   2838             city: Some("c".to_string()),
   2839             region: Some("r".to_string()),
   2840             country: Some("k".to_string()),
   2841             geohash: "9q8yy".to_string(),
   2842         };
   2843         assert_eq!(
   2844             unpack_farm_location_strings(Some(&location)).0,
   2845             Some("p".to_string())
   2846         );
   2847         assert_eq!(
   2848             unpack_plot_location_strings(Some(&PlotLocation {
   2849                 primary: Some("p".to_string()),
   2850                 city: None,
   2851                 region: None,
   2852                 country: None,
   2853                 gcs: sample_gcs(13.0, 23.0, "s3"),
   2854             }))
   2855             .0,
   2856             Some("p".to_string())
   2857         );
   2858         assert!(ensure_list_set_entries_tag(&bad_image, "p", "x").is_ok());
   2859         assert!(
   2860             ensure_list_set_entries_tag(
   2861                 &ListSet {
   2862                     d_tag: "x".to_string(),
   2863                     content: String::new(),
   2864                     entries: vec![ListEntry {
   2865                         tag: "a".to_string(),
   2866                         values: vec!["x".to_string()],
   2867                     }],
   2868                     title: None,
   2869                     description: None,
   2870                     image: None,
   2871                 },
   2872                 "p",
   2873                 "x",
   2874             )
   2875             .is_err()
   2876         );
   2877     }
   2878 
   2879     #[test]
   2880     fn ingest_listing_projects_trade_product_and_removes_archived_replacements() {
   2881         let exec = SqlxSqliteExecutor::open_memory().expect("db");
   2882         migrations::run_all_up(&exec).expect("migrations");
   2883 
   2884         let seller_pubkey = FIXTURE_ALICE_PUBLIC_KEY_HEX.to_owned();
   2885         let listing_d_tag = "AAAAAAAAAAAAAAAAAAAAAQ";
   2886         let listing_addr = format!(
   2887             "{}:{}:{}",
   2888             KIND_CLASSIFIED_LISTING, seller_pubkey, listing_d_tag
   2889         );
   2890 
   2891         let mut active = listing_event(&seller_pubkey, 10, listing_d_tag, "active", "Pasture Eggs");
   2892         let mut active_tags = active.tags_as_vec();
   2893         active_tags.push(vec![
   2894             "radroots:discount".to_string(),
   2895             serde_json::json!({
   2896                 "scope": "bin",
   2897                 "threshold": {
   2898                     "kind": "bin_count",
   2899                     "amount": { "bin_id": "bin-a", "min": 1 }
   2900                 },
   2901                 "value": {
   2902                     "kind": "percent",
   2903                     "amount": { "value": "10" }
   2904                 }
   2905             })
   2906             .to_string(),
   2907         ]);
   2908         active = signed_listing_event(
   2909             &seller_pubkey,
   2910             active.created_at_u64(),
   2911             active_tags,
   2912             active.content().to_owned(),
   2913         );
   2914         assert_eq!(
   2915             radroots_replica_ingest_event(&exec, &active).expect("active ingest"),
   2916             RadrootsReplicaIngestOutcome::Applied
   2917         );
   2918 
   2919         let replica = ReplicaSql::new(&exec);
   2920         let search_rows = replica
   2921             .trade_product_search(&["eggs".to_string()])
   2922             .expect("search");
   2923         assert_eq!(search_rows.len(), 1);
   2924         assert_eq!(
   2925             search_rows[0].listing_addr.as_deref(),
   2926             Some(listing_addr.as_str())
   2927         );
   2928         assert_eq!(search_rows[0].title, "Pasture Eggs");
   2929         assert_eq!(search_rows[0].primary_bin_id.as_deref(), Some("bin-a"));
   2930         assert_eq!(
   2931             search_rows[0].verified_primary_bin_id.as_deref(),
   2932             Some("bin-a")
   2933         );
   2934         assert_eq!(search_rows[0].qty_amt, 12.0);
   2935         assert_eq!(search_rows[0].qty_amt_exact.as_deref(), Some("12"));
   2936         assert_eq!(search_rows[0].qty_avail, Some(5));
   2937         assert_eq!(search_rows[0].price_amt, 6.0);
   2938         assert_eq!(search_rows[0].price_amt_exact.as_deref(), Some("6"));
   2939         assert_eq!(search_rows[0].price_currency, "USD");
   2940         assert_eq!(search_rows[0].price_qty_amt, 1.0);
   2941         assert_eq!(search_rows[0].price_qty_amt_exact.as_deref(), Some("1"));
   2942         assert!(
   2943             search_rows[0]
   2944                 .notes
   2945                 .as_deref()
   2946                 .is_some_and(|notes| notes.contains("listing_discounts"))
   2947         );
   2948 
   2949         let updated = listing_event(&seller_pubkey, 11, listing_d_tag, "active", "Market Eggs");
   2950         assert_eq!(
   2951             radroots_replica_ingest_event(&exec, &updated).expect("listing update"),
   2952             RadrootsReplicaIngestOutcome::Applied
   2953         );
   2954         let product_rows = trade_product::find_many(
   2955             &exec,
   2956             &ITradeProductFindMany {
   2957                 filter: Some(trade_product_listing_addr_filter(&listing_addr)),
   2958             },
   2959         )
   2960         .expect("product rows")
   2961         .results;
   2962         assert_eq!(product_rows.len(), 1);
   2963         assert_eq!(product_rows[0].title, "Market Eggs");
   2964         assert_eq!(product_rows[0].primary_bin_id.as_deref(), Some("bin-a"));
   2965         assert_eq!(
   2966             product_rows[0].verified_primary_bin_id.as_deref(),
   2967             Some("bin-a")
   2968         );
   2969 
   2970         let archived = listing_event(&seller_pubkey, 12, listing_d_tag, "archived", "Market Eggs");
   2971         assert_eq!(
   2972             radroots_replica_ingest_event(&exec, &archived).expect("archived ingest"),
   2973             RadrootsReplicaIngestOutcome::Applied
   2974         );
   2975         let search_rows = replica
   2976             .trade_product_search(&["eggs".to_string()])
   2977             .expect("search archived");
   2978         assert!(search_rows.is_empty());
   2979 
   2980         let product_rows = trade_product::find_many(
   2981             &exec,
   2982             &ITradeProductFindMany {
   2983                 filter: Some(trade_product_listing_addr_filter(&listing_addr)),
   2984             },
   2985         )
   2986         .expect("archived product rows")
   2987         .results;
   2988         assert!(product_rows.is_empty());
   2989 
   2990         let state = nostr_event_head::find_one(
   2991             &exec,
   2992             &INostrEventHeadFindOne::On(INostrEventHeadFindOneArgs {
   2993                 on: NostrEventHeadQueryBindValues::Key {
   2994                     key: event_head_key(KIND_CLASSIFIED_LISTING, &seller_pubkey, listing_d_tag),
   2995                 },
   2996             }),
   2997         )
   2998         .expect("event state")
   2999         .result
   3000         .expect("state row");
   3001         assert_eq!(state.last_event_id, archived.id_hex());
   3002 
   3003         let stale_active = listing_event(&seller_pubkey, 11, listing_d_tag, "active", "Stale Eggs");
   3004         assert_eq!(
   3005             radroots_replica_ingest_event(&exec, &stale_active).expect("stale ingest"),
   3006             RadrootsReplicaIngestOutcome::Skipped
   3007         );
   3008         let product_rows = trade_product::find_many(
   3009             &exec,
   3010             &ITradeProductFindMany {
   3011                 filter: Some(trade_product_listing_addr_filter(&listing_addr)),
   3012             },
   3013         )
   3014         .expect("stale product rows")
   3015         .results;
   3016         assert!(product_rows.is_empty());
   3017     }
   3018 
   3019     #[test]
   3020     fn ingest_listing_preserves_fractional_exact_economics() {
   3021         let exec = SqlxSqliteExecutor::open_memory().expect("db");
   3022         migrations::run_all_up(&exec).expect("migrations");
   3023 
   3024         let seller_pubkey = FIXTURE_ALICE_PUBLIC_KEY_HEX.to_owned();
   3025         let listing_d_tag = "AAAAAAAAAAAAAAAAAAAAAg";
   3026         let listing_addr = format!(
   3027             "{}:{}:{}",
   3028             KIND_CLASSIFIED_LISTING, seller_pubkey, listing_d_tag
   3029         );
   3030 
   3031         let mut active = listing_event(
   3032             &seller_pubkey,
   3033             10,
   3034             listing_d_tag,
   3035             "active",
   3036             "Half Gram Greens",
   3037         );
   3038         let mut active_tags = active.tags_as_vec();
   3039         for tag in &mut active_tags {
   3040             if tag.first().is_some_and(|name| name == "radroots:bin") {
   3041                 tag[2] = "0.5".to_string();
   3042                 tag[3] = "g".to_string();
   3043                 tag[4] = "0.5".to_string();
   3044                 tag[5] = "g".to_string();
   3045                 tag[6] = "half gram".to_string();
   3046             }
   3047             if tag.first().is_some_and(|name| name == "radroots:price") {
   3048                 tag[2] = "3.25".to_string();
   3049                 tag[3] = "USD".to_string();
   3050                 tag[4] = "1".to_string();
   3051                 tag[5] = "g".to_string();
   3052                 tag[6] = "3.25".to_string();
   3053                 tag[7] = "g".to_string();
   3054             }
   3055         }
   3056         active = signed_listing_event(
   3057             &seller_pubkey,
   3058             active.created_at_u64(),
   3059             active_tags,
   3060             active.content().to_owned(),
   3061         );
   3062 
   3063         assert_eq!(
   3064             radroots_replica_ingest_event(&exec, &active).expect("fractional active ingest"),
   3065             RadrootsReplicaIngestOutcome::Applied
   3066         );
   3067 
   3068         let replica = ReplicaSql::new(&exec);
   3069         let search_rows = replica
   3070             .trade_product_search(&["greens".to_string()])
   3071             .expect("search");
   3072         assert_eq!(search_rows.len(), 1);
   3073         assert_eq!(
   3074             search_rows[0].listing_addr.as_deref(),
   3075             Some(listing_addr.as_str())
   3076         );
   3077         assert_eq!(search_rows[0].qty_amt, 0.5);
   3078         assert_eq!(search_rows[0].qty_amt_exact.as_deref(), Some("0.5"));
   3079         assert_eq!(search_rows[0].price_amt, 3.25);
   3080         assert_eq!(search_rows[0].price_amt_exact.as_deref(), Some("3.25"));
   3081         assert_eq!(search_rows[0].price_qty_amt, 1.0);
   3082         assert_eq!(search_rows[0].price_qty_amt_exact.as_deref(), Some("1"));
   3083     }
   3084 
   3085     #[test]
   3086     fn listing_raw_head_partition_prevents_projection_fallback_and_tampering() {
   3087         let exec = SqlxSqliteExecutor::open_memory().expect("db");
   3088         migrations::run_all_up(&exec).expect("migrations");
   3089 
   3090         let seller = FIXTURE_ALICE_PUBLIC_KEY_HEX;
   3091         let d_tag = "AAAAAAAAAAAAAAAAAAAAAw";
   3092         let listing_addr = event_head_key(KIND_CLASSIFIED_LISTING, seller, d_tag);
   3093         let product_count = || {
   3094             trade_product::find_many(
   3095                 &exec,
   3096                 &ITradeProductFindMany {
   3097                     filter: Some(trade_product_listing_addr_filter(&listing_addr)),
   3098                 },
   3099             )
   3100             .expect("listing products")
   3101             .results
   3102             .len()
   3103         };
   3104 
   3105         let active = listing_event(seller, 10, d_tag, "active", "Pasture Eggs");
   3106         assert_eq!(
   3107             radroots_replica_ingest_event(&exec, &active).expect("operational ingest"),
   3108             RadrootsReplicaIngestOutcome::Applied
   3109         );
   3110         assert_eq!(product_count(), 1);
   3111 
   3112         let focused = focused_listing_event(seller, 20, d_tag);
   3113         assert_eq!(
   3114             radroots_replica_ingest_event(&exec, &focused).expect("focused ingest"),
   3115             RadrootsReplicaIngestOutcome::Excluded
   3116         );
   3117         assert_eq!(product_count(), 0);
   3118 
   3119         let invalid_focused = signed_listing_event(
   3120             seller,
   3121             25,
   3122             vec![
   3123                 vec!["d".to_string(), d_tag.to_string()],
   3124                 vec!["radroots:price_unit".to_string(), "lb".to_string()],
   3125             ],
   3126             "invalid focused listing".to_string(),
   3127         );
   3128         assert_eq!(
   3129             radroots_replica_ingest_event(&exec, &invalid_focused).expect("invalid focused ingest"),
   3130             RadrootsReplicaIngestOutcome::Rejected
   3131         );
   3132         assert_eq!(product_count(), 0);
   3133 
   3134         let stale = listing_event(seller, 15, d_tag, "active", "Stale Eggs");
   3135         assert_eq!(
   3136             radroots_replica_ingest_event(&exec, &stale).expect("stale ingest"),
   3137             RadrootsReplicaIngestOutcome::Skipped
   3138         );
   3139         assert_eq!(product_count(), 0);
   3140 
   3141         let active = listing_event(seller, 30, d_tag, "active", "Market Eggs");
   3142         assert_eq!(
   3143             radroots_replica_ingest_event(&exec, &active).expect("operational replacement"),
   3144             RadrootsReplicaIngestOutcome::Applied
   3145         );
   3146         assert_eq!(product_count(), 1);
   3147 
   3148         let generic = generic_listing_event(seller, 40, d_tag);
   3149         assert_eq!(
   3150             radroots_replica_ingest_event(&exec, &generic).expect("generic ingest"),
   3151             RadrootsReplicaIngestOutcome::Excluded
   3152         );
   3153         assert_eq!(product_count(), 0);
   3154 
   3155         let active = listing_event(seller, 50, d_tag, "active", "Market Eggs");
   3156         assert_eq!(
   3157             radroots_replica_ingest_event(&exec, &active).expect("operational replacement"),
   3158             RadrootsReplicaIngestOutcome::Applied
   3159         );
   3160         let ambiguous = signed_listing_event(
   3161             seller,
   3162             60,
   3163             vec![
   3164                 vec!["d".to_string(), d_tag.to_string()],
   3165                 vec!["radroots:price_unit".to_string(), "lb".to_string()],
   3166                 vec!["radroots:primary_bin".to_string(), "bin-a".to_string()],
   3167             ],
   3168             "ambiguous".to_string(),
   3169         );
   3170         assert_eq!(
   3171             radroots_replica_ingest_event(&exec, &ambiguous).expect("ambiguous ingest"),
   3172             RadrootsReplicaIngestOutcome::Rejected
   3173         );
   3174         assert_eq!(product_count(), 0);
   3175 
   3176         let active = listing_event(seller, 70, d_tag, "active", "Market Eggs");
   3177         assert_eq!(
   3178             radroots_replica_ingest_event(&exec, &active).expect("operational replacement"),
   3179             RadrootsReplicaIngestOutcome::Applied
   3180         );
   3181         let malformed_operational = signed_listing_event(
   3182             seller,
   3183             80,
   3184             vec![
   3185                 vec!["d".to_string(), d_tag.to_string()],
   3186                 vec!["radroots:primary_bin".to_string()],
   3187             ],
   3188             "malformed".to_string(),
   3189         );
   3190         assert_eq!(
   3191             radroots_replica_ingest_event(&exec, &malformed_operational)
   3192                 .expect("malformed operational ingest"),
   3193             RadrootsReplicaIngestOutcome::Rejected
   3194         );
   3195         assert_eq!(product_count(), 0);
   3196 
   3197         let active = listing_event(seller, 90, d_tag, "active", "Market Eggs");
   3198         assert_eq!(
   3199             radroots_replica_ingest_event(&exec, &active).expect("operational replacement"),
   3200             RadrootsReplicaIngestOutcome::Applied
   3201         );
   3202         assert_eq!(product_count(), 1);
   3203 
   3204         let signed_replacement = focused_listing_event(seller, 100, d_tag);
   3205         assert!(matches!(
   3206             radroots_replica_ingest_event_head(&exec, &signed_replacement),
   3207             Err(RadrootsReplicaEventsError::InvalidData(ref message))
   3208                 if message == "classified listing heads require profile-aware replica ingestion"
   3209         ));
   3210         assert_eq!(product_count(), 1);
   3211 
   3212         let tampered = test_event_with_content(&signed_replacement, "tampered".to_string());
   3213         assert!(matches!(
   3214             radroots_replica_ingest_event(&exec, &tampered),
   3215             Err(RadrootsReplicaEventsError::Verification(_))
   3216         ));
   3217         assert!(matches!(
   3218             radroots_replica_ingest_event_head(&exec, &tampered),
   3219             Err(RadrootsReplicaEventsError::Verification(_))
   3220         ));
   3221         assert_eq!(product_count(), 1);
   3222 
   3223         let state_before_replacement = nostr_event_head::find_one(
   3224             &exec,
   3225             &INostrEventHeadFindOne::On(INostrEventHeadFindOneArgs {
   3226                 on: NostrEventHeadQueryBindValues::Key {
   3227                     key: listing_addr.clone(),
   3228                 },
   3229             }),
   3230         )
   3231         .expect("event state")
   3232         .result
   3233         .expect("state row");
   3234         assert_eq!(state_before_replacement.last_event_id, active.id_hex());
   3235 
   3236         assert_eq!(
   3237             radroots_replica_ingest_event(&exec, &signed_replacement)
   3238                 .expect("profile-aware focused replacement"),
   3239             RadrootsReplicaIngestOutcome::Excluded
   3240         );
   3241         assert_eq!(product_count(), 0);
   3242 
   3243         let tied_operational =
   3244             listing_event(seller, 110, d_tag, "active", "Equal-time Market Eggs");
   3245         assert_eq!(
   3246             radroots_replica_ingest_event(&exec, &tied_operational)
   3247                 .expect("equal-time operational head"),
   3248             RadrootsReplicaIngestOutcome::Applied
   3249         );
   3250         assert_eq!(product_count(), 1);
   3251 
   3252         let tied_focused = (0..256)
   3253             .map(|nonce| {
   3254                 focused_listing_event_with_content(
   3255                     seller,
   3256                     110,
   3257                     d_tag,
   3258                     &format!("Equal-time focused replacement {nonce}"),
   3259                 )
   3260             })
   3261             .find(|candidate| candidate.id_hex() < tied_operational.id_hex())
   3262             .expect("deterministic lower-id focused candidate");
   3263         assert_eq!(
   3264             radroots_replica_ingest_event(&exec, &tied_focused)
   3265                 .expect("lower-id focused replacement"),
   3266             RadrootsReplicaIngestOutcome::Excluded
   3267         );
   3268         assert_eq!(product_count(), 0);
   3269         assert_eq!(
   3270             radroots_replica_ingest_event(&exec, &tied_operational)
   3271                 .expect("higher-id equal-time operational replay"),
   3272             RadrootsReplicaIngestOutcome::Skipped
   3273         );
   3274 
   3275         let final_state = nostr_event_head::find_one(
   3276             &exec,
   3277             &INostrEventHeadFindOne::On(INostrEventHeadFindOneArgs {
   3278                 on: NostrEventHeadQueryBindValues::Key { key: listing_addr },
   3279             }),
   3280         )
   3281         .expect("final event state")
   3282         .result
   3283         .expect("final state row");
   3284         assert_eq!(final_state.last_event_id, tied_focused.id_hex());
   3285     }
   3286 
   3287     #[test]
   3288     fn upsert_location_none_paths_are_ok() {
   3289         let exec = SqlxSqliteExecutor::open_memory().expect("db");
   3290         migrations::run_all_up(&exec).expect("migrations");
   3291 
   3292         let farm_row = farm::create(
   3293             &exec,
   3294             &IFarmFields {
   3295                 d_tag: "AAAAAAAAAAAAAAAAAAAAAA".to_string(),
   3296                 pubkey: fixture_public_key(15),
   3297                 name: "farm-none".to_string(),
   3298                 about: None,
   3299                 website: None,
   3300                 picture: None,
   3301                 banner: None,
   3302                 location_primary: None,
   3303                 location_city: None,
   3304                 location_region: None,
   3305                 location_country: None,
   3306             },
   3307         )
   3308         .expect("farm")
   3309         .result;
   3310         let plot_row = plot::create(
   3311             &exec,
   3312             &IPlotFields {
   3313                 d_tag: "AAAAAAAAAAAAAAAAAAAAAQ".to_string(),
   3314                 farm_id: farm_row.id.clone(),
   3315                 name: "plot-none".to_string(),
   3316                 about: None,
   3317                 location_primary: None,
   3318                 location_city: None,
   3319                 location_region: None,
   3320                 location_country: None,
   3321             },
   3322         )
   3323         .expect("plot")
   3324         .result;
   3325 
   3326         upsert_plot_location(&exec, &plot_row.id, None, &FixedFactory).expect("plot none");
   3327     }
   3328 
   3329     #[test]
   3330     fn ingest_delete_error_paths_are_covered() {
   3331         let exec = SqlxSqliteExecutor::open_memory().expect("db");
   3332         let (farm_id, _farm_pubkey, farm_d_tag, _plot_d_tag) = seed_rows(&exec);
   3333 
   3334         let not_found_farm_tags = DeleteErrorExecutor {
   3335             inner: &exec,
   3336             table_name: "farm_tag",
   3337             err: SqlError::NotFound("farm_tag".to_string()),
   3338         };
   3339         assert!(
   3340             upsert_farm_tags(
   3341                 &not_found_farm_tags,
   3342                 &farm_id,
   3343                 Some(vec!["next".to_string()])
   3344             )
   3345             .is_ok()
   3346         );
   3347 
   3348         let not_found_plot_tags = DeleteErrorExecutor {
   3349             inner: &exec,
   3350             table_name: "plot_tag",
   3351             err: SqlError::NotFound("plot_tag".to_string()),
   3352         };
   3353         let plot_id = plot::find_many(&exec, &IPlotFindMany { filter: None })
   3354             .expect("plots")
   3355             .results[0]
   3356             .id
   3357             .clone();
   3358         assert!(
   3359             upsert_plot_tags(
   3360                 &not_found_plot_tags,
   3361                 &plot_id,
   3362                 Some(vec!["next".to_string()])
   3363             )
   3364             .is_ok()
   3365         );
   3366         assert!(
   3367             upsert_plot_tags(
   3368                 &exec,
   3369                 &plot_id,
   3370                 Some(vec!["next".to_string(), " ".to_string()])
   3371             )
   3372             .is_ok()
   3373         );
   3374 
   3375         let not_found_plot_locations = DeleteErrorExecutor {
   3376             inner: &exec,
   3377             table_name: "plot_gcs_location",
   3378             err: SqlError::NotFound("plot_gcs_location".to_string()),
   3379         };
   3380         assert!(
   3381             upsert_plot_location(
   3382                 &not_found_plot_locations,
   3383                 &plot_id,
   3384                 Some(PlotLocation {
   3385                     primary: None,
   3386                     city: None,
   3387                     region: None,
   3388                     country: None,
   3389                     gcs: sample_gcs(1.1, 2.1, "s5"),
   3390                 }),
   3391                 &FixedFactory,
   3392             )
   3393             .is_ok()
   3394         );
   3395 
   3396         let members_list_set =
   3397             farm_list_sets::farm_members_list_set(&farm_d_tag, vec![fixture_public_key(7)])
   3398                 .expect("members");
   3399         assert!(
   3400             upsert_farm_members(&exec, &farm_id, ListSetRole::Members, &members_list_set).is_ok()
   3401         );
   3402         let not_found_members = DeleteErrorExecutor {
   3403             inner: &exec,
   3404             table_name: "farm_member",
   3405             err: SqlError::NotFound("farm_member".to_string()),
   3406         };
   3407         let not_found_members_list_set =
   3408             farm_list_sets::farm_members_list_set(&farm_d_tag, vec![fixture_public_key(10)])
   3409                 .expect("not found members");
   3410         assert!(
   3411             upsert_farm_members(
   3412                 &not_found_members,
   3413                 &farm_id,
   3414                 ListSetRole::Members,
   3415                 &not_found_members_list_set,
   3416             )
   3417             .is_ok()
   3418         );
   3419         assert!(
   3420             upsert_farm_members(
   3421                 &not_found_members,
   3422                 &farm_id,
   3423                 ListSetRole::Plots,
   3424                 &not_found_members_list_set,
   3425             )
   3426             .is_ok()
   3427         );
   3428 
   3429         let member_claims =
   3430             farm_list_sets::member_of_farms_list_set(vec![fixture_public_key(3)]).expect("claims");
   3431         assert!(upsert_member_claims(&exec, &fixture_public_key(6), &member_claims).is_ok());
   3432         let not_found_claims = DeleteErrorExecutor {
   3433             inner: &exec,
   3434             table_name: "farm_member_claim",
   3435             err: SqlError::NotFound("farm_member_claim".to_string()),
   3436         };
   3437         let not_found_member_claims =
   3438             farm_list_sets::member_of_farms_list_set(vec![fixture_public_key(2)])
   3439                 .expect("claims nf");
   3440         assert!(
   3441             upsert_member_claims(
   3442                 &not_found_claims,
   3443                 &fixture_public_key(6),
   3444                 &not_found_member_claims
   3445             )
   3446             .is_ok()
   3447         );
   3448         assert!(not_found_claims.begin().is_ok());
   3449         assert!(not_found_claims.commit().is_ok());
   3450         let _ = not_found_claims.rollback();
   3451         assert!(not_found_claims.query_raw("SELECT 1", "[]").is_ok());
   3452         assert_eq!(
   3453             not_found_claims
   3454                 .exec("DELETE FROM farm_member_claim WHERE id = 1", "[]")
   3455                 .expect_err("exec not found")
   3456                 .code(),
   3457             "ERR_NOT_FOUND"
   3458         );
   3459         let _ = not_found_claims.exec("DELETE FROM other_table WHERE id = 1", "[]");
   3460 
   3461         let internal_farm_tags = DeleteErrorExecutor {
   3462             inner: &exec,
   3463             table_name: "farm_tag",
   3464             err: SqlError::Internal,
   3465         };
   3466         assert!(
   3467             upsert_farm_tags(&internal_farm_tags, &farm_id, Some(vec!["x".to_string()])).is_err()
   3468         );
   3469 
   3470         let internal_plot_tags = DeleteErrorExecutor {
   3471             inner: &exec,
   3472             table_name: "plot_tag",
   3473             err: SqlError::Internal,
   3474         };
   3475         assert!(
   3476             upsert_plot_tags(&internal_plot_tags, &plot_id, Some(vec!["x".to_string()])).is_err()
   3477         );
   3478 
   3479         let internal_plot_locations = DeleteErrorExecutor {
   3480             inner: &exec,
   3481             table_name: "plot_gcs_location",
   3482             err: SqlError::Internal,
   3483         };
   3484         assert!(
   3485             upsert_plot_location(
   3486                 &internal_plot_locations,
   3487                 &plot_id,
   3488                 Some(PlotLocation {
   3489                     primary: None,
   3490                     city: None,
   3491                     region: None,
   3492                     country: None,
   3493                     gcs: sample_gcs(2.1, 3.1, "s7"),
   3494                 }),
   3495                 &FixedFactory,
   3496             )
   3497             .is_err()
   3498         );
   3499 
   3500         let internal_members = DeleteErrorExecutor {
   3501             inner: &exec,
   3502             table_name: "farm_member",
   3503             err: SqlError::Internal,
   3504         };
   3505         assert!(
   3506             upsert_farm_members(
   3507                 &internal_members,
   3508                 &farm_id,
   3509                 ListSetRole::Members,
   3510                 &members_list_set,
   3511             )
   3512             .is_err()
   3513         );
   3514 
   3515         let internal_claims = DeleteErrorExecutor {
   3516             inner: &exec,
   3517             table_name: "farm_member_claim",
   3518             err: SqlError::Internal,
   3519         };
   3520         assert!(
   3521             upsert_member_claims(&internal_claims, &fixture_public_key(6), &member_claims).is_err()
   3522         );
   3523     }
   3524 
   3525     #[test]
   3526     fn create_gcs_location_error_mapping_helpers_are_covered() {
   3527         let point_json_err = serde_json::from_str::<Value>("{").expect_err("invalid json");
   3528         let point_err = map_gcs_point_serialize_error(point_json_err);
   3529         assert_eq!(point_err.to_string(), "replica_sync.data: gcs.point");
   3530 
   3531         let polygon_json_err = serde_json::from_str::<Value>("{").expect_err("invalid json");
   3532         let polygon_err = map_gcs_polygon_serialize_error(polygon_json_err);
   3533         assert_eq!(polygon_err.to_string(), "replica_sync.data: gcs.polygon");
   3534     }
   3535 
   3536     #[test]
   3537     fn ingest_pass_executor_and_parse_edge_paths_are_covered() {
   3538         let exec = SqlxSqliteExecutor::open_memory().expect("db");
   3539         migrations::run_all_up(&exec).expect("migrations");
   3540         let pass = PassExecutor { inner: &exec };
   3541 
   3542         let profile_pubkey = fixture_public_key(9);
   3543         let farm_pubkey = fixture_public_key(15);
   3544         let farm_d_tag = "AAAAAAAAAAAAAAAAAAAAAA";
   3545         let plot_d_tag = "AAAAAAAAAAAAAAAAAAAAAQ";
   3546 
   3547         let profile = profile_event(
   3548             500,
   3549             &profile_pubkey,
   3550             50,
   3551             Some(ProfileType::Individual),
   3552             "pass-profile",
   3553         );
   3554         let signed_profile = sign_test_event(&profile);
   3555         assert_eq!(
   3556             radroots_replica_ingest_event_with_factory(&pass, &signed_profile, &FixedFactory)
   3557                 .expect("profile ingest"),
   3558             RadrootsReplicaIngestOutcome::Applied
   3559         );
   3560         assert_eq!(
   3561             ingest_test_profile(&pass, &profile).expect("profile skip"),
   3562             RadrootsReplicaIngestOutcome::Skipped
   3563         );
   3564 
   3565         let farm = farm_event(
   3566             501,
   3567             &farm_pubkey,
   3568             51,
   3569             farm_d_tag,
   3570             "pass-farm",
   3571             Some(FarmPublicLocation {
   3572                 primary: "primary".to_string(),
   3573                 city: Some("city".to_string()),
   3574                 region: Some("region".to_string()),
   3575                 country: Some("country".to_string()),
   3576                 geohash: "9q8yy".to_string(),
   3577             }),
   3578             Some(vec!["coffee".to_string(), "coffee".to_string()]),
   3579         );
   3580         assert_eq!(
   3581             ingest_farm_event(&pass, &farm, &FixedFactory).expect("farm ingest"),
   3582             RadrootsReplicaIngestOutcome::Applied
   3583         );
   3584 
   3585         let plot = plot_event(
   3586             502,
   3587             &farm_pubkey,
   3588             52,
   3589             plot_d_tag,
   3590             FarmRef {
   3591                 pubkey: farm_pubkey.clone(),
   3592                 d_tag: farm_d_tag.to_string(),
   3593             },
   3594             "pass-plot",
   3595             Some(PlotLocation {
   3596                 primary: Some("plot".to_string()),
   3597                 city: None,
   3598                 region: None,
   3599                 country: None,
   3600                 gcs: sample_gcs(11.0, 21.0, "s1"),
   3601             }),
   3602             Some(vec!["orchard".to_string()]),
   3603         );
   3604         assert_eq!(
   3605             ingest_plot_event(&pass, &plot, &FixedFactory).expect("plot ingest"),
   3606             RadrootsReplicaIngestOutcome::Applied
   3607         );
   3608 
   3609         let members =
   3610             farm_list_sets::farm_members_list_set(farm_d_tag, vec![fixture_public_key(6)])
   3611                 .expect("list");
   3612         let members_event = list_set_event(503, &farm_pubkey, 53, KIND_LIST_SET_GENERIC, &members);
   3613         assert_eq!(
   3614             ingest_list_set_event(&pass, &members_event).expect("members list set"),
   3615             RadrootsReplicaIngestOutcome::Applied
   3616         );
   3617 
   3618         let claims = farm_list_sets::member_of_farms_list_set(vec![farm_pubkey.clone()])
   3619             .expect("claims list set");
   3620         let claims_event = list_set_event(504, &profile_pubkey, 54, KIND_LIST_SET_GENERIC, &claims);
   3621         assert_eq!(
   3622             ingest_list_set_event(&pass, &claims_event).expect("claims list set"),
   3623             RadrootsReplicaIngestOutcome::Applied
   3624         );
   3625 
   3626         let farm_row = find_farm_by_ref(&pass, &farm_pubkey, farm_d_tag).expect("farm row");
   3627         let mixed_member_entries = ListSet {
   3628             d_tag: format!("farm:{farm_d_tag}:members"),
   3629             content: String::new(),
   3630             entries: vec![
   3631                 ListEntry {
   3632                     tag: "p".to_string(),
   3633                     values: Vec::new(),
   3634                 },
   3635                 ListEntry {
   3636                     tag: "p".to_string(),
   3637                     values: vec![fixture_public_key(6)],
   3638                 },
   3639             ],
   3640             title: None,
   3641             description: None,
   3642             image: None,
   3643         };
   3644         assert!(
   3645             upsert_farm_members(
   3646                 &pass,
   3647                 &farm_row.id,
   3648                 ListSetRole::Members,
   3649                 &mixed_member_entries
   3650             )
   3651             .is_ok()
   3652         );
   3653         let mixed_claim_entries = ListSet {
   3654             d_tag: "member_of.farms".to_string(),
   3655             content: String::new(),
   3656             entries: vec![
   3657                 ListEntry {
   3658                     tag: "p".to_string(),
   3659                     values: Vec::new(),
   3660                 },
   3661                 ListEntry {
   3662                     tag: "p".to_string(),
   3663                     values: vec![farm_pubkey.clone()],
   3664                 },
   3665             ],
   3666             title: None,
   3667             description: None,
   3668             image: None,
   3669         };
   3670         assert!(upsert_member_claims(&pass, &profile_pubkey, &mixed_claim_entries).is_ok());
   3671         assert!(pass.begin().is_ok());
   3672         assert!(pass.rollback().is_ok());
   3673 
   3674         assert!(parse_farm_list_set_d_tag("coop:AAAAAAAAAAAAAAAAAAAAAA:members").is_none());
   3675         assert!(parse_farm_list_set_d_tag("farm:AAAAAAAAAAAAAAAAAAAAAA").is_none());
   3676         assert!(parse_farm_list_set_d_tag("farm:AAAAAAAAAAAAAAAAAAAAAA:members").is_some());
   3677     }
   3678 
   3679     #[test]
   3680     fn create_gcs_location_success_path_is_covered() {
   3681         let exec = SqlxSqliteExecutor::open_memory().expect("db");
   3682         migrations::run_all_up(&exec).expect("migrations");
   3683 
   3684         let id = create_gcs_location(&exec, sample_gcs(1.0, 2.0, "s0"), &FixedFactory)
   3685             .expect("create gcs");
   3686         assert!(!id.trim().is_empty());
   3687     }
   3688 
   3689     #[test]
   3690     fn ingest_default_factory_wrapper_paths_are_covered() {
   3691         let exec = SqlxSqliteExecutor::open_memory().expect("db");
   3692         migrations::run_all_up(&exec).expect("migrations");
   3693 
   3694         let farm_pubkey = fixture_public_key(15);
   3695         let farm_d_tag = "AAAAAAAAAAAAAAAAAAAAAA";
   3696         let farm_create = farm_event(
   3697             600,
   3698             &farm_pubkey,
   3699             60,
   3700             farm_d_tag,
   3701             "wrapper-farm",
   3702             Some(FarmPublicLocation {
   3703                 primary: "primary".to_string(),
   3704                 city: Some("city".to_string()),
   3705                 region: None,
   3706                 country: None,
   3707                 geohash: "9q8yy".to_string(),
   3708             }),
   3709             Some(vec!["coffee".to_string()]),
   3710         );
   3711         assert_eq!(
   3712             radroots_replica_ingest_event(&exec, &farm_create).expect("farm create"),
   3713             RadrootsReplicaIngestOutcome::Applied
   3714         );
   3715 
   3716         let farm_update = farm_event(
   3717             601,
   3718             &farm_pubkey,
   3719             61,
   3720             farm_d_tag,
   3721             "wrapper-farm-updated",
   3722             None,
   3723             Some(vec!["market".to_string()]),
   3724         );
   3725         assert_eq!(
   3726             radroots_replica_ingest_event(&exec, &farm_update).expect("farm update"),
   3727             RadrootsReplicaIngestOutcome::Applied
   3728         );
   3729 
   3730         let plot_d_tag = "AAAAAAAAAAAAAAAAAAAAAQ";
   3731         let plot_create = plot_event(
   3732             602,
   3733             &farm_pubkey,
   3734             62,
   3735             plot_d_tag,
   3736             FarmRef {
   3737                 pubkey: farm_pubkey.clone(),
   3738                 d_tag: farm_d_tag.to_string(),
   3739             },
   3740             "wrapper-plot",
   3741             Some(PlotLocation {
   3742                 primary: Some("plot-primary".to_string()),
   3743                 city: None,
   3744                 region: None,
   3745                 country: None,
   3746                 gcs: sample_gcs(11.0, 21.0, "s1"),
   3747             }),
   3748             Some(vec!["orchard".to_string()]),
   3749         );
   3750         assert_eq!(
   3751             radroots_replica_ingest_event(&exec, &plot_create).expect("plot create"),
   3752             RadrootsReplicaIngestOutcome::Applied
   3753         );
   3754 
   3755         let plot_update = plot_event(
   3756             603,
   3757             &farm_pubkey,
   3758             63,
   3759             plot_d_tag,
   3760             FarmRef {
   3761                 pubkey: farm_pubkey.clone(),
   3762                 d_tag: farm_d_tag.to_string(),
   3763             },
   3764             "wrapper-plot-updated",
   3765             None,
   3766             Some(vec!["updated".to_string()]),
   3767         );
   3768         assert_eq!(
   3769             radroots_replica_ingest_event(&exec, &plot_update).expect("plot update"),
   3770             RadrootsReplicaIngestOutcome::Applied
   3771         );
   3772     }
   3773 
   3774     #[test]
   3775     fn ingest_txn_executor_instantiation_error_paths_are_covered() {
   3776         let pass_db = SqlxSqliteExecutor::open_memory().expect("db");
   3777         migrations::run_all_up(&pass_db).expect("migrations");
   3778         let pass_txn = TxnExecutor {
   3779             inner: Some(&pass_db),
   3780             begin_err: None,
   3781             commit_err: None,
   3782             rollback_count: Arc::new(AtomicUsize::new(0)),
   3783         };
   3784 
   3785         let profile_pubkey = fixture_public_key(9);
   3786         let profile_event_row = profile_event(
   3787             700,
   3788             &profile_pubkey,
   3789             70,
   3790             Some(ProfileType::Individual),
   3791             "txn-profile",
   3792         );
   3793         assert_eq!(
   3794             ingest_test_profile(&pass_txn, &profile_event_row).expect("txn profile"),
   3795             RadrootsReplicaIngestOutcome::Applied
   3796         );
   3797         let signed_profile_event_row = sign_test_event(&profile_event_row);
   3798         let profile_decision =
   3799             event_head_decision(&pass_txn, &signed_profile_event_row).expect("profile decision");
   3800         assert!(!profile_decision.apply);
   3801         assert!(radroots_replica_ingest_event_head(&pass_txn, &signed_profile_event_row).is_ok());
   3802         assert_eq!(
   3803             radroots_replica_ingest_event_with_factory(
   3804                 &pass_txn,
   3805                 &signed_profile_event_row,
   3806                 &FixedFactory
   3807             )
   3808             .expect("txn wrapper"),
   3809             RadrootsReplicaIngestOutcome::Skipped
   3810         );
   3811 
   3812         let farm_pubkey = fixture_public_key(15);
   3813         let farm_d_tag = "AAAAAAAAAAAAAAAAAAAAAA";
   3814         let farm_event_row = farm_event(
   3815             701,
   3816             &farm_pubkey,
   3817             71,
   3818             farm_d_tag,
   3819             "txn-farm",
   3820             Some(FarmPublicLocation {
   3821                 primary: "primary".to_string(),
   3822                 city: Some("city".to_string()),
   3823                 region: None,
   3824                 country: None,
   3825                 geohash: "9q8yy".to_string(),
   3826             }),
   3827             Some(vec!["coffee".to_string()]),
   3828         );
   3829         assert_eq!(
   3830             ingest_farm_event(&pass_txn, &farm_event_row, &FixedFactory).expect("txn farm"),
   3831             RadrootsReplicaIngestOutcome::Applied
   3832         );
   3833 
   3834         let plot_event_row = plot_event(
   3835             702,
   3836             &farm_pubkey,
   3837             72,
   3838             "AAAAAAAAAAAAAAAAAAAAAQ",
   3839             FarmRef {
   3840                 pubkey: farm_pubkey.clone(),
   3841                 d_tag: farm_d_tag.to_string(),
   3842             },
   3843             "txn-plot",
   3844             Some(PlotLocation {
   3845                 primary: Some("primary".to_string()),
   3846                 city: None,
   3847                 region: None,
   3848                 country: None,
   3849                 gcs: sample_gcs(13.0, 23.0, "s3"),
   3850             }),
   3851             Some(vec!["orchard".to_string()]),
   3852         );
   3853         assert_eq!(
   3854             ingest_plot_event(&pass_txn, &plot_event_row, &FixedFactory).expect("txn plot"),
   3855             RadrootsReplicaIngestOutcome::Applied
   3856         );
   3857 
   3858         let farm_row = find_farm_by_ref(&pass_txn, &farm_pubkey, farm_d_tag).expect("find farm");
   3859         assert!(upsert_farm_tags(&pass_txn, &farm_row.id, Some(vec!["x".to_string()])).is_ok());
   3860         let plot_id = plot::find_many(&pass_db, &IPlotFindMany { filter: None })
   3861             .expect("plots")
   3862             .results[0]
   3863             .id
   3864             .clone();
   3865         assert!(upsert_plot_tags(&pass_txn, &plot_id, Some(vec!["y".to_string()])).is_ok());
   3866         assert!(clear_farm_locations(&pass_txn, &farm_row.id).is_ok());
   3867         assert!(clear_plot_locations(&pass_txn, &plot_id).is_ok());
   3868         assert!(
   3869             create_gcs_location(&pass_txn, sample_gcs(14.0, 24.0, "s4"), &FixedFactory).is_ok()
   3870         );
   3871         assert!(
   3872             upsert_plot_location(
   3873                 &pass_txn,
   3874                 &plot_id,
   3875                 Some(PlotLocation {
   3876                     primary: Some("primary".to_string()),
   3877                     city: None,
   3878                     region: None,
   3879                     country: None,
   3880                     gcs: sample_gcs(16.0, 26.0, "s6"),
   3881                 }),
   3882                 &FixedFactory,
   3883             )
   3884             .is_ok()
   3885         );
   3886         let members_list =
   3887             farm_list_sets::farm_members_list_set(farm_d_tag, vec![fixture_public_key(6)])
   3888                 .expect("list");
   3889         assert!(
   3890             upsert_farm_members(&pass_txn, &farm_row.id, ListSetRole::Members, &members_list)
   3891                 .is_ok()
   3892         );
   3893         let member_of_list =
   3894             farm_list_sets::member_of_farms_list_set(vec![farm_pubkey.clone()]).expect("member_of");
   3895         assert!(upsert_member_claims(&pass_txn, &fixture_public_key(6), &member_of_list).is_ok());
   3896 
   3897         let rollback_count = Arc::new(AtomicUsize::new(0));
   3898         let txn = TxnExecutor {
   3899             inner: None,
   3900             begin_err: None,
   3901             commit_err: None,
   3902             rollback_count,
   3903         };
   3904 
   3905         assert!(ingest_test_profile(&txn, &profile_event_row).is_err());
   3906         assert!(event_head_decision(&txn, &profile_event_row).is_err());
   3907         assert!(radroots_replica_ingest_event_head(&txn, &profile_event_row).is_err());
   3908         let signed_profile_event_row = sign_test_event(&profile_event_row);
   3909         assert!(
   3910             radroots_replica_ingest_event_with_factory(
   3911                 &txn,
   3912                 &signed_profile_event_row,
   3913                 &FixedFactory
   3914             )
   3915             .is_err()
   3916         );
   3917 
   3918         assert!(ingest_farm_event(&txn, &farm_event_row, &FixedFactory).is_err());
   3919 
   3920         assert!(ingest_plot_event(&txn, &plot_event_row, &FixedFactory).is_err());
   3921 
   3922         assert!(find_farm_by_ref(&txn, &farm_pubkey, farm_d_tag).is_err());
   3923         assert!(upsert_farm_tags(&txn, "farm-id", Some(vec!["x".to_string()])).is_err());
   3924         assert!(upsert_plot_tags(&txn, "plot-id", Some(vec!["y".to_string()])).is_err());
   3925         assert!(clear_farm_locations(&txn, "farm-id").is_err());
   3926         assert!(clear_plot_locations(&txn, "plot-id").is_err());
   3927         assert!(create_gcs_location(&txn, sample_gcs(14.0, 24.0, "s4"), &FixedFactory).is_err());
   3928         assert!(
   3929             upsert_plot_location(
   3930                 &txn,
   3931                 "plot-id",
   3932                 Some(PlotLocation {
   3933                     primary: Some("primary".to_string()),
   3934                     city: None,
   3935                     region: None,
   3936                     country: None,
   3937                     gcs: sample_gcs(16.0, 26.0, "s6"),
   3938                 }),
   3939                 &FixedFactory,
   3940             )
   3941             .is_err()
   3942         );
   3943         assert!(upsert_farm_members(&txn, "farm-id", ListSetRole::Members, &members_list).is_err());
   3944         assert!(upsert_member_claims(&txn, &fixture_public_key(6), &member_of_list).is_err());
   3945     }
   3946 
   3947     #[test]
   3948     fn ingest_sqlite_queryfail_and_parser_edges_are_covered() {
   3949         let exec = SqlxSqliteExecutor::open_memory().expect("db");
   3950         migrations::run_all_up(&exec).expect("migrations");
   3951         let pass_through = QueryFailExecutor {
   3952             inner: &exec,
   3953             needle: "__missing__",
   3954             err: SqlError::Internal,
   3955         };
   3956         assert!(pass_through.query_raw("select 1", "[]").is_ok());
   3957         assert!(
   3958             pass_through
   3959                 .exec(
   3960                     "create table if not exists coverage_probe (id integer)",
   3961                     "[]"
   3962                 )
   3963                 .is_ok()
   3964         );
   3965         let _ = pass_through.begin();
   3966         let _ = pass_through.rollback();
   3967         let _ = pass_through.commit();
   3968 
   3969         let farm_pubkey = fixture_public_key(15);
   3970         let farm_d_tag = "AAAAAAAAAAAAAAAAAAAAAA";
   3971         let plot_d_tag = "AAAAAAAAAAAAAAAAAAAAAQ";
   3972         let profile_pubkey = fixture_public_key(9);
   3973 
   3974         let profile = profile_event(
   3975             800,
   3976             &profile_pubkey,
   3977             80,
   3978             Some(ProfileType::Individual),
   3979             "profile-base",
   3980         );
   3981         let profile_bad_content = test_event_with_content(&profile, "{".to_string());
   3982         assert!(ingest_test_profile(&exec, &profile_bad_content).is_err());
   3983 
   3984         let profile_query_fail = QueryFailExecutor {
   3985             inner: &exec,
   3986             needle: "nostr_profile",
   3987             err: SqlError::Internal,
   3988         };
   3989         assert!(ingest_test_profile(&profile_query_fail, &profile).is_err());
   3990 
   3991         assert_eq!(
   3992             ingest_test_profile(&exec, &profile).expect("profile seed"),
   3993             RadrootsReplicaIngestOutcome::Applied
   3994         );
   3995         let profile_update = profile_event(
   3996             801,
   3997             &profile_pubkey,
   3998             81,
   3999             Some(ProfileType::Individual),
   4000             "profile-update",
   4001         );
   4002         let profile_update_fail = QueryFailExecutor {
   4003             inner: &exec,
   4004             needle: "update nostr_profile",
   4005             err: SqlError::Internal,
   4006         };
   4007         assert!(ingest_test_profile(&profile_update_fail, &profile_update).is_err());
   4008 
   4009         let profile_create_fail = QueryFailExecutor {
   4010             inner: &exec,
   4011             needle: "insert into nostr_profile",
   4012             err: SqlError::Internal,
   4013         };
   4014         let profile_new = profile_event(
   4015             802,
   4016             &fixture_public_key(7),
   4017             82,
   4018             Some(ProfileType::Individual),
   4019             "profile-new",
   4020         );
   4021         assert!(ingest_test_profile(&profile_create_fail, &profile_new).is_err());
   4022 
   4023         let profile_state_fail = QueryFailExecutor {
   4024             inner: &exec,
   4025             needle: "nostr_event_head",
   4026             err: SqlError::Internal,
   4027         };
   4028         let profile_state_event = profile_event(
   4029             803,
   4030             &fixture_public_key(12),
   4031             83,
   4032             Some(ProfileType::Individual),
   4033             "profile-state",
   4034         );
   4035         assert!(ingest_test_profile(&profile_state_fail, &profile_state_event).is_err());
   4036 
   4037         let farm_seed = farm_event(
   4038             810,
   4039             &farm_pubkey,
   4040             90,
   4041             farm_d_tag,
   4042             "farm-seed",
   4043             Some(FarmPublicLocation {
   4044                 primary: "primary".to_string(),
   4045                 city: Some("city".to_string()),
   4046                 region: Some("region".to_string()),
   4047                 country: Some("country".to_string()),
   4048                 geohash: "9q8yy".to_string(),
   4049             }),
   4050             Some(vec!["seed".to_string()]),
   4051         );
   4052         assert_eq!(
   4053             ingest_farm_event(&exec, &farm_seed, &FixedFactory).expect("farm seed"),
   4054             RadrootsReplicaIngestOutcome::Applied
   4055         );
   4056 
   4057         let farm_bad_content = test_event_with_content(&farm_seed, "{".to_string());
   4058         assert!(ingest_farm_event(&exec, &farm_bad_content, &FixedFactory).is_err());
   4059 
   4060         let farm_query_fail = QueryFailExecutor {
   4061             inner: &exec,
   4062             needle: "from farm",
   4063             err: SqlError::Internal,
   4064         };
   4065         let farm_query_event = farm_event(
   4066             811,
   4067             &fixture_public_key(10),
   4068             91,
   4069             farm_d_tag,
   4070             "farm-query",
   4071             None,
   4072             None,
   4073         );
   4074         assert!(ingest_farm_event(&farm_query_fail, &farm_query_event, &FixedFactory).is_err());
   4075 
   4076         let farm_update_fail = QueryFailExecutor {
   4077             inner: &exec,
   4078             needle: "update farm",
   4079             err: SqlError::Internal,
   4080         };
   4081         let farm_update = farm_event(
   4082             812,
   4083             &farm_pubkey,
   4084             92,
   4085             farm_d_tag,
   4086             "farm-update",
   4087             None,
   4088             Some(vec!["u".to_string()]),
   4089         );
   4090         assert!(ingest_farm_event(&farm_update_fail, &farm_update, &FixedFactory).is_err());
   4091 
   4092         let farm_create_fail = QueryFailExecutor {
   4093             inner: &exec,
   4094             needle: "insert into farm",
   4095             err: SqlError::Internal,
   4096         };
   4097         let farm_create = farm_event(
   4098             813,
   4099             &fixture_public_key(12),
   4100             93,
   4101             farm_d_tag,
   4102             "farm-create",
   4103             None,
   4104             None,
   4105         );
   4106         assert!(ingest_farm_event(&farm_create_fail, &farm_create, &FixedFactory).is_err());
   4107 
   4108         let farm_tag_fail = QueryFailExecutor {
   4109             inner: &exec,
   4110             needle: "farm_tag",
   4111             err: SqlError::Internal,
   4112         };
   4113         let farm_tag_event = farm_event(
   4114             814,
   4115             &fixture_public_key(13),
   4116             94,
   4117             farm_d_tag,
   4118             "farm-tag",
   4119             None,
   4120             Some(vec!["coffee".to_string()]),
   4121         );
   4122         assert!(ingest_farm_event(&farm_tag_fail, &farm_tag_event, &FixedFactory).is_err());
   4123 
   4124         let farm_gcs_fail = QueryFailExecutor {
   4125             inner: &exec,
   4126             needle: "gcs_location",
   4127             err: SqlError::Internal,
   4128         };
   4129         let farm_gcs_event = farm_event(
   4130             815,
   4131             &fixture_public_key(16),
   4132             95,
   4133             farm_d_tag,
   4134             "farm-gcs",
   4135             Some(FarmPublicLocation {
   4136                 primary: "primary".to_string(),
   4137                 city: Some("city".to_string()),
   4138                 region: None,
   4139                 country: None,
   4140                 geohash: "9q8yy".to_string(),
   4141             }),
   4142             None,
   4143         );
   4144         assert_eq!(
   4145             ingest_farm_event(&farm_gcs_fail, &farm_gcs_event, &FixedFactory).expect("farm gcs"),
   4146             RadrootsReplicaIngestOutcome::Applied
   4147         );
   4148 
   4149         let farm_rel_fail = QueryFailExecutor {
   4150             inner: &exec,
   4151             needle: "farm_gcs_location",
   4152             err: SqlError::Internal,
   4153         };
   4154         let farm_rel_event = farm_event(
   4155             816,
   4156             &fixture_public_key(11),
   4157             96,
   4158             farm_d_tag,
   4159             "farm-rel",
   4160             Some(FarmPublicLocation {
   4161                 primary: "primary".to_string(),
   4162                 city: Some("city".to_string()),
   4163                 region: None,
   4164                 country: None,
   4165                 geohash: "9q8yy".to_string(),
   4166             }),
   4167             None,
   4168         );
   4169         assert_eq!(
   4170             ingest_farm_event(&farm_rel_fail, &farm_rel_event, &FixedFactory).expect("farm rel"),
   4171             RadrootsReplicaIngestOutcome::Applied
   4172         );
   4173 
   4174         let farm_state_fail = QueryFailExecutor {
   4175             inner: &exec,
   4176             needle: "nostr_event_head",
   4177             err: SqlError::Internal,
   4178         };
   4179         let farm_state_event = farm_event(
   4180             817,
   4181             &fixture_public_key(16),
   4182             97,
   4183             farm_d_tag,
   4184             "farm-state",
   4185             None,
   4186             None,
   4187         );
   4188         assert!(ingest_farm_event(&farm_state_fail, &farm_state_event, &FixedFactory).is_err());
   4189 
   4190         let farm_public_location = farm_event(
   4191             818,
   4192             &fixture_public_key(1),
   4193             98,
   4194             farm_d_tag,
   4195             "farm-public-location",
   4196             Some(FarmPublicLocation {
   4197                 primary: "primary".to_string(),
   4198                 city: Some("city".to_string()),
   4199                 region: None,
   4200                 country: None,
   4201                 geohash: "9q8yy".to_string(),
   4202             }),
   4203             None,
   4204         );
   4205         assert_eq!(
   4206             ingest_farm_event(&exec, &farm_public_location, &FixedFactory)
   4207                 .expect("farm public location"),
   4208             RadrootsReplicaIngestOutcome::Applied
   4209         );
   4210 
   4211         let farm_public_locality = farm_event(
   4212             819,
   4213             &fixture_public_key(2),
   4214             99,
   4215             farm_d_tag,
   4216             "farm-public-locality",
   4217             Some(FarmPublicLocation {
   4218                 primary: "primary".to_string(),
   4219                 city: Some("city".to_string()),
   4220                 region: None,
   4221                 country: None,
   4222                 geohash: "9q8yy".to_string(),
   4223             }),
   4224             None,
   4225         );
   4226         assert_eq!(
   4227             ingest_farm_event(&exec, &farm_public_locality, &FixedFactory)
   4228                 .expect("farm public locality"),
   4229             RadrootsReplicaIngestOutcome::Applied
   4230         );
   4231 
   4232         let plot_seed = plot_event(
   4233             820,
   4234             &farm_pubkey,
   4235             100,
   4236             plot_d_tag,
   4237             FarmRef {
   4238                 pubkey: farm_pubkey.clone(),
   4239                 d_tag: farm_d_tag.to_string(),
   4240             },
   4241             "plot-seed",
   4242             Some(PlotLocation {
   4243                 primary: Some("primary".to_string()),
   4244                 city: None,
   4245                 region: None,
   4246                 country: None,
   4247                 gcs: sample_gcs(15.0, 25.0, "s5"),
   4248             }),
   4249             Some(vec!["orchard".to_string()]),
   4250         );
   4251         assert_eq!(
   4252             ingest_plot_event(&exec, &plot_seed, &FixedFactory).expect("plot seed"),
   4253             RadrootsReplicaIngestOutcome::Applied
   4254         );
   4255 
   4256         let plot_bad_content = test_event_with_content(&plot_seed, "{".to_string());
   4257         assert!(ingest_plot_event(&exec, &plot_bad_content, &FixedFactory).is_err());
   4258 
   4259         let plot_query_fail = QueryFailExecutor {
   4260             inner: &exec,
   4261             needle: "from plot",
   4262             err: SqlError::Internal,
   4263         };
   4264         let plot_query = plot_event(
   4265             821,
   4266             &farm_pubkey,
   4267             101,
   4268             plot_d_tag,
   4269             FarmRef {
   4270                 pubkey: farm_pubkey.clone(),
   4271                 d_tag: farm_d_tag.to_string(),
   4272             },
   4273             "plot-query",
   4274             None,
   4275             None,
   4276         );
   4277         assert!(ingest_plot_event(&plot_query_fail, &plot_query, &FixedFactory).is_err());
   4278 
   4279         let plot_update_fail = QueryFailExecutor {
   4280             inner: &exec,
   4281             needle: "update plot",
   4282             err: SqlError::Internal,
   4283         };
   4284         let plot_update = plot_event(
   4285             822,
   4286             &farm_pubkey,
   4287             102,
   4288             plot_d_tag,
   4289             FarmRef {
   4290                 pubkey: farm_pubkey.clone(),
   4291                 d_tag: farm_d_tag.to_string(),
   4292             },
   4293             "plot-update",
   4294             None,
   4295             Some(vec!["u".to_string()]),
   4296         );
   4297         assert!(ingest_plot_event(&plot_update_fail, &plot_update, &FixedFactory).is_err());
   4298 
   4299         let plot_create_fail = QueryFailExecutor {
   4300             inner: &exec,
   4301             needle: "insert into plot",
   4302             err: SqlError::Internal,
   4303         };
   4304         let plot_create = plot_event(
   4305             823,
   4306             &farm_pubkey,
   4307             103,
   4308             "AAAAAAAAAAAAAAAAAAAAAg",
   4309             FarmRef {
   4310                 pubkey: farm_pubkey.clone(),
   4311                 d_tag: farm_d_tag.to_string(),
   4312             },
   4313             "plot-create",
   4314             None,
   4315             None,
   4316         );
   4317         assert!(ingest_plot_event(&plot_create_fail, &plot_create, &FixedFactory).is_err());
   4318 
   4319         let plot_tag_fail = QueryFailExecutor {
   4320             inner: &exec,
   4321             needle: "plot_tag",
   4322             err: SqlError::Internal,
   4323         };
   4324         let plot_tag_event = plot_event(
   4325             824,
   4326             &farm_pubkey,
   4327             104,
   4328             "AAAAAAAAAAAAAAAAAAAAAw",
   4329             FarmRef {
   4330                 pubkey: farm_pubkey.clone(),
   4331                 d_tag: farm_d_tag.to_string(),
   4332             },
   4333             "plot-tag",
   4334             None,
   4335             Some(vec!["tag".to_string()]),
   4336         );
   4337         assert!(ingest_plot_event(&plot_tag_fail, &plot_tag_event, &FixedFactory).is_err());
   4338 
   4339         let plot_gcs_fail = QueryFailExecutor {
   4340             inner: &exec,
   4341             needle: "gcs_location",
   4342             err: SqlError::Internal,
   4343         };
   4344         let plot_gcs_event = plot_event(
   4345             825,
   4346             &farm_pubkey,
   4347             105,
   4348             "AAAAAAAAAAAAAAAAAAAAAw",
   4349             FarmRef {
   4350                 pubkey: farm_pubkey.clone(),
   4351                 d_tag: farm_d_tag.to_string(),
   4352             },
   4353             "plot-gcs",
   4354             Some(PlotLocation {
   4355                 primary: Some("primary".to_string()),
   4356                 city: None,
   4357                 region: None,
   4358                 country: None,
   4359                 gcs: sample_gcs(16.0, 26.0, "s6"),
   4360             }),
   4361             None,
   4362         );
   4363         assert!(ingest_plot_event(&plot_gcs_fail, &plot_gcs_event, &FixedFactory).is_err());
   4364 
   4365         let plot_rel_fail = QueryFailExecutor {
   4366             inner: &exec,
   4367             needle: "plot_gcs_location",
   4368             err: SqlError::Internal,
   4369         };
   4370         let plot_rel_event = plot_event(
   4371             826,
   4372             &farm_pubkey,
   4373             106,
   4374             "AAAAAAAAAAAAAAAAAAAAAw",
   4375             FarmRef {
   4376                 pubkey: farm_pubkey.clone(),
   4377                 d_tag: farm_d_tag.to_string(),
   4378             },
   4379             "plot-rel",
   4380             Some(PlotLocation {
   4381                 primary: Some("primary".to_string()),
   4382                 city: None,
   4383                 region: None,
   4384                 country: None,
   4385                 gcs: sample_gcs(17.0, 27.0, "s7"),
   4386             }),
   4387             None,
   4388         );
   4389         assert!(ingest_plot_event(&plot_rel_fail, &plot_rel_event, &FixedFactory).is_err());
   4390 
   4391         let plot_state_fail = QueryFailExecutor {
   4392             inner: &exec,
   4393             needle: "nostr_event_head",
   4394             err: SqlError::Internal,
   4395         };
   4396         let plot_state_event = plot_event(
   4397             827,
   4398             &farm_pubkey,
   4399             107,
   4400             "AAAAAAAAAAAAAAAAAAAAAw",
   4401             FarmRef {
   4402                 pubkey: farm_pubkey.clone(),
   4403                 d_tag: farm_d_tag.to_string(),
   4404             },
   4405             "plot-state",
   4406             None,
   4407             None,
   4408         );
   4409         assert!(ingest_plot_event(&plot_state_fail, &plot_state_event, &FixedFactory).is_err());
   4410 
   4411         let list_decode_fail =
   4412             profile_event(830, &farm_pubkey, 108, Some(ProfileType::Farm), "unused");
   4413         let list_decode_fail = test_event_with_parts(
   4414             &list_decode_fail,
   4415             KIND_LIST_SET_GENERIC,
   4416             Vec::new(),
   4417             "{".to_string(),
   4418         );
   4419         assert!(ingest_list_set_event(&exec, &list_decode_fail).is_err());
   4420 
   4421         let members_list =
   4422             farm_list_sets::farm_members_list_set(farm_d_tag, vec![fixture_public_key(6)])
   4423                 .expect("members list");
   4424         let member_event =
   4425             list_set_event(831, &farm_pubkey, 109, KIND_LIST_SET_GENERIC, &members_list);
   4426         let list_decision_fail = QueryFailExecutor {
   4427             inner: &exec,
   4428             needle: "nostr_event_head",
   4429             err: SqlError::Internal,
   4430         };
   4431         assert!(ingest_list_set_event(&list_decision_fail, &member_event).is_err());
   4432 
   4433         let member_of =
   4434             farm_list_sets::member_of_farms_list_set(vec![farm_pubkey.clone()]).expect("member-of");
   4435         let member_of_event = list_set_event(
   4436             832,
   4437             &fixture_public_key(6),
   4438             110,
   4439             KIND_LIST_SET_GENERIC,
   4440             &member_of,
   4441         );
   4442         let claims_fail = QueryFailExecutor {
   4443             inner: &exec,
   4444             needle: "farm_member_claim",
   4445             err: SqlError::Internal,
   4446         };
   4447         assert!(ingest_list_set_event(&claims_fail, &member_of_event).is_err());
   4448 
   4449         let claims_state_fail = QueryFailExecutor {
   4450             inner: &exec,
   4451             needle: "nostr_event_head",
   4452             err: SqlError::Internal,
   4453         };
   4454         assert!(ingest_list_set_event(&claims_state_fail, &member_of_event).is_err());
   4455 
   4456         let plots_list = farm_list_sets::farm_plots_list_set(
   4457             farm_d_tag,
   4458             &farm_pubkey,
   4459             vec![plot_d_tag.to_string()],
   4460         )
   4461         .expect("plots list");
   4462         let plots_event =
   4463             list_set_event(833, &farm_pubkey, 111, KIND_LIST_SET_GENERIC, &plots_list);
   4464         let plots_state_fail = QueryFailExecutor {
   4465             inner: &exec,
   4466             needle: "nostr_event_head",
   4467             err: SqlError::Internal,
   4468         };
   4469         assert!(ingest_list_set_event(&plots_state_fail, &plots_event).is_err());
   4470 
   4471         let missing_farm_members =
   4472             farm_list_sets::farm_members_list_set(farm_d_tag, vec![fixture_public_key(7)])
   4473                 .expect("list");
   4474         let missing_farm_event = list_set_event(
   4475             834,
   4476             &fixture_public_key(3),
   4477             112,
   4478             KIND_LIST_SET_GENERIC,
   4479             &missing_farm_members,
   4480         );
   4481         assert!(ingest_list_set_event(&exec, &missing_farm_event).is_err());
   4482 
   4483         let members_create_fail = QueryFailExecutor {
   4484             inner: &exec,
   4485             needle: "farm_member",
   4486             err: SqlError::Internal,
   4487         };
   4488         assert!(ingest_list_set_event(&members_create_fail, &member_event).is_err());
   4489 
   4490         let members_state_fail = QueryFailExecutor {
   4491             inner: &exec,
   4492             needle: "nostr_event_head",
   4493             err: SqlError::Internal,
   4494         };
   4495         assert!(ingest_list_set_event(&members_state_fail, &member_event).is_err());
   4496 
   4497         assert!(parse_farm_list_set_d_tag("").is_none());
   4498         assert!(parse_farm_list_set_d_tag("farm").is_none());
   4499 
   4500         let state_create_fail = QueryFailExecutor {
   4501             inner: &exec,
   4502             needle: "nostr_event_head",
   4503             err: SqlError::Internal,
   4504         };
   4505         assert!(radroots_replica_ingest_event_head(&state_create_fail, &profile).is_err());
   4506 
   4507         radroots_replica_ingest_event_head(&exec, &profile).expect("seed state");
   4508         let state_update_fail = QueryFailExecutor {
   4509             inner: &exec,
   4510             needle: "update nostr_event_head",
   4511             err: SqlError::Internal,
   4512         };
   4513         let profile_update = profile_event(
   4514             808,
   4515             &fixture_public_key(9),
   4516             101,
   4517             Some(ProfileType::Individual),
   4518             "profile-update-error",
   4519         );
   4520         assert!(radroots_replica_ingest_event_head(&state_update_fail, &profile_update).is_err());
   4521     }
   4522 
   4523     #[test]
   4524     fn ingest_insert_and_state_error_branches_are_covered() {
   4525         let exec = SqlxSqliteExecutor::open_memory().expect("db");
   4526         let (farm_id, farm_pubkey, farm_d_tag, plot_d_tag) = seed_rows(&exec);
   4527 
   4528         let profile = profile_event(
   4529             900,
   4530             &fixture_public_key(14),
   4531             120,
   4532             Some(ProfileType::Individual),
   4533             "profile-state-insert",
   4534         );
   4535         let state_insert_fail = QueryFailExecutor {
   4536             inner: &exec,
   4537             needle: "insert into nostr_event_head",
   4538             err: SqlError::Internal,
   4539         };
   4540         assert!(ingest_test_profile(&state_insert_fail, &profile).is_err());
   4541 
   4542         let farm_state = farm_event(
   4543             901,
   4544             &fixture_public_key(10),
   4545             121,
   4546             "AAAAAAAAAAAAAAAAAAAAAQ",
   4547             "farm-state-insert",
   4548             None,
   4549             None,
   4550         );
   4551         assert!(ingest_farm_event(&state_insert_fail, &farm_state, &FixedFactory).is_err());
   4552 
   4553         let plot_state = plot_event(
   4554             902,
   4555             &farm_pubkey,
   4556             122,
   4557             "AAAAAAAAAAAAAAAAAAAAAg",
   4558             FarmRef {
   4559                 pubkey: farm_pubkey.clone(),
   4560                 d_tag: farm_d_tag.clone(),
   4561             },
   4562             "plot-state-insert",
   4563             None,
   4564             None,
   4565         );
   4566         assert!(ingest_plot_event(&state_insert_fail, &plot_state, &FixedFactory).is_err());
   4567 
   4568         let members_set =
   4569             farm_list_sets::farm_members_list_set(&farm_d_tag, vec![fixture_public_key(7)])
   4570                 .expect("members");
   4571         let members_event =
   4572             list_set_event(903, &farm_pubkey, 123, KIND_LIST_SET_GENERIC, &members_set);
   4573         assert!(ingest_list_set_event(&state_insert_fail, &members_event).is_err());
   4574 
   4575         let plots_set = farm_list_sets::farm_plots_list_set(
   4576             &farm_d_tag,
   4577             &farm_pubkey,
   4578             vec![plot_d_tag.clone()],
   4579         )
   4580         .expect("plots");
   4581         let plots_event = list_set_event(904, &farm_pubkey, 124, KIND_LIST_SET_GENERIC, &plots_set);
   4582         assert!(ingest_list_set_event(&state_insert_fail, &plots_event).is_err());
   4583 
   4584         let member_of_set =
   4585             farm_list_sets::member_of_farms_list_set(vec![farm_pubkey.clone()]).expect("member_of");
   4586         let member_of_event = list_set_event(
   4587             905,
   4588             &fixture_public_key(7),
   4589             125,
   4590             KIND_LIST_SET_GENERIC,
   4591             &member_of_set,
   4592         );
   4593         assert!(ingest_list_set_event(&state_insert_fail, &member_of_event).is_err());
   4594 
   4595         let state_insert_only_fail = QueryFailExecutor {
   4596             inner: &exec,
   4597             needle: "insert into nostr_event_head",
   4598             err: SqlError::Internal,
   4599         };
   4600         assert!(radroots_replica_ingest_event_head(&state_insert_only_fail, &profile).is_err());
   4601 
   4602         crate::event_head::event_content_hash_fail_next();
   4603         assert!(event_head_decision(&exec, &profile).is_err());
   4604 
   4605         let farm_tag_insert_fail = QueryFailExecutor {
   4606             inner: &exec,
   4607             needle: "insert into farm_tag",
   4608             err: SqlError::Internal,
   4609         };
   4610         assert!(
   4611             upsert_farm_tags(
   4612                 &farm_tag_insert_fail,
   4613                 &farm_id,
   4614                 Some(vec!["delta".to_string()])
   4615             )
   4616             .is_err()
   4617         );
   4618 
   4619         let plot_id = plot::find_many(&exec, &IPlotFindMany { filter: None })
   4620             .expect("plots")
   4621             .results[0]
   4622             .id
   4623             .clone();
   4624         let plot_tag_insert_fail = QueryFailExecutor {
   4625             inner: &exec,
   4626             needle: "insert into plot_tag",
   4627             err: SqlError::Internal,
   4628         };
   4629         assert!(
   4630             upsert_plot_tags(
   4631                 &plot_tag_insert_fail,
   4632                 &plot_id,
   4633                 Some(vec!["epsilon".to_string()])
   4634             )
   4635             .is_err()
   4636         );
   4637 
   4638         let plot_gcs_insert_fail = QueryFailExecutor {
   4639             inner: &exec,
   4640             needle: "insert into gcs_location",
   4641             err: SqlError::Internal,
   4642         };
   4643         assert!(
   4644             upsert_plot_location(
   4645                 &plot_gcs_insert_fail,
   4646                 &plot_id,
   4647                 Some(PlotLocation {
   4648                     primary: Some("primary".to_string()),
   4649                     city: None,
   4650                     region: None,
   4651                     country: None,
   4652                     gcs: sample_gcs(33.0, 43.0, "sa"),
   4653                 }),
   4654                 &FixedFactory,
   4655             )
   4656             .is_err()
   4657         );
   4658 
   4659         let plot_rel_insert_fail = QueryFailExecutor {
   4660             inner: &exec,
   4661             needle: "insert into plot_gcs_location",
   4662             err: SqlError::Internal,
   4663         };
   4664         assert!(
   4665             upsert_plot_location(
   4666                 &plot_rel_insert_fail,
   4667                 &plot_id,
   4668                 Some(PlotLocation {
   4669                     primary: Some("primary".to_string()),
   4670                     city: None,
   4671                     region: None,
   4672                     country: None,
   4673                     gcs: sample_gcs(34.0, 44.0, "sb"),
   4674                 }),
   4675                 &FixedFactory,
   4676             )
   4677             .is_err()
   4678         );
   4679 
   4680         let member_insert_fail = QueryFailExecutor {
   4681             inner: &exec,
   4682             needle: "insert into farm_member",
   4683             err: SqlError::Internal,
   4684         };
   4685         assert!(
   4686             upsert_farm_members(
   4687                 &member_insert_fail,
   4688                 &farm_id,
   4689                 ListSetRole::Members,
   4690                 &members_set
   4691             )
   4692             .is_err()
   4693         );
   4694 
   4695         let claims_insert_fail = QueryFailExecutor {
   4696             inner: &exec,
   4697             needle: "insert into farm_member_claim",
   4698             err: SqlError::Internal,
   4699         };
   4700         assert!(
   4701             upsert_member_claims(&claims_insert_fail, &fixture_public_key(7), &member_of_set)
   4702                 .is_err()
   4703         );
   4704 
   4705         super::failpoints::set_gcs_point_serialize_error();
   4706         assert!(create_gcs_location(&exec, sample_gcs(35.0, 45.0, "sc"), &FixedFactory).is_err());
   4707 
   4708         super::failpoints::set_gcs_polygon_serialize_error();
   4709         assert!(create_gcs_location(&exec, sample_gcs(36.0, 46.0, "sd"), &FixedFactory).is_err());
   4710 
   4711         assert!(parse_farm_list_set_d_tag("coop:AAAAAAAAAAAAAAAAAAAAAA:members").is_none());
   4712     }
   4713 
   4714     #[test]
   4715     fn upsert_member_helpers_ignore_empty_entry_values() {
   4716         let exec = SqlxSqliteExecutor::open_memory().expect("db");
   4717         let (farm_id, farm_pubkey, _, _) = seed_rows(&exec);
   4718 
   4719         let member_pubkey = fixture_public_key(6);
   4720         let member_list_set = ListSet {
   4721             d_tag: "farm:AAAAAAAAAAAAAAAAAAAAAQ:members".to_string(),
   4722             content: String::new(),
   4723             entries: vec![
   4724                 ListEntry {
   4725                     tag: "p".to_string(),
   4726                     values: Vec::new(),
   4727                 },
   4728                 ListEntry {
   4729                     tag: "p".to_string(),
   4730                     values: vec![member_pubkey.clone(), "ignored".to_string()],
   4731                 },
   4732                 ListEntry {
   4733                     tag: "p".to_string(),
   4734                     values: vec![member_pubkey.clone()],
   4735                 },
   4736             ],
   4737             title: None,
   4738             description: None,
   4739             image: None,
   4740         };
   4741         upsert_farm_members(&exec, &farm_id, ListSetRole::Members, &member_list_set)
   4742             .expect("members");
   4743         let members = farm_member::find_many(
   4744             &exec,
   4745             &IFarmMemberFindMany {
   4746                 filter: Some(IFarmMemberFieldsFilter {
   4747                     id: None,
   4748                     created_at: None,
   4749                     updated_at: None,
   4750                     farm_id: Some(farm_id.clone()),
   4751                     member_pubkey: None,
   4752                     role: Some(ROLE_MEMBER.to_string()),
   4753                 }),
   4754             },
   4755         )
   4756         .expect("member rows")
   4757         .results;
   4758         assert_eq!(members.len(), 1);
   4759         assert_eq!(members[0].member_pubkey, member_pubkey);
   4760 
   4761         upsert_farm_members(&exec, &farm_id, ListSetRole::Plots, &member_list_set)
   4762             .expect("plots is no-op");
   4763 
   4764         let claimant_pubkey = fixture_public_key(7);
   4765         let claims_list_set = ListSet {
   4766             d_tag: "member_of.farms".to_string(),
   4767             content: String::new(),
   4768             entries: vec![
   4769                 ListEntry {
   4770                     tag: "p".to_string(),
   4771                     values: Vec::new(),
   4772                 },
   4773                 ListEntry {
   4774                     tag: "p".to_string(),
   4775                     values: vec![farm_pubkey.clone(), "ignored".to_string()],
   4776                 },
   4777                 ListEntry {
   4778                     tag: "p".to_string(),
   4779                     values: vec![farm_pubkey.clone()],
   4780                 },
   4781             ],
   4782             title: None,
   4783             description: None,
   4784             image: None,
   4785         };
   4786         upsert_member_claims(&exec, &claimant_pubkey, &claims_list_set).expect("claims");
   4787         let claims = farm_member_claim::find_many(
   4788             &exec,
   4789             &IFarmMemberClaimFindMany {
   4790                 filter: Some(IFarmMemberClaimFieldsFilter {
   4791                     id: None,
   4792                     created_at: None,
   4793                     updated_at: None,
   4794                     member_pubkey: Some(claimant_pubkey),
   4795                     farm_pubkey: None,
   4796                 }),
   4797             },
   4798         )
   4799         .expect("claim rows")
   4800         .results;
   4801         assert_eq!(claims.len(), 1);
   4802         assert_eq!(claims[0].farm_pubkey, farm_pubkey);
   4803     }
   4804 
   4805     #[test]
   4806     fn ingest_error_paths_cover_missing_farm_and_bad_list_set_tags() {
   4807         let exec = SqlxSqliteExecutor::open_memory().expect("db");
   4808         let (_, farm_pubkey, farm_d_tag, plot_d_tag) = seed_rows(&exec);
   4809 
   4810         let missing_farm_plot = plot_event(
   4811             950,
   4812             &farm_pubkey,
   4813             220,
   4814             &plot_d_tag,
   4815             FarmRef {
   4816                 pubkey: fixture_public_key(1),
   4817                 d_tag: farm_d_tag.clone(),
   4818             },
   4819             "plot-missing-farm",
   4820             None,
   4821             None,
   4822         );
   4823         assert!(ingest_plot_event(&exec, &missing_farm_plot, &FixedFactory).is_err());
   4824 
   4825         let mut bad_member_of =
   4826             farm_list_sets::member_of_farms_list_set(vec![farm_pubkey.clone()]).expect("member_of");
   4827         bad_member_of.entries[0].tag = "x".to_string();
   4828         let bad_member_of_event = list_set_event(
   4829             951,
   4830             &fixture_public_key(7),
   4831             221,
   4832             KIND_LIST_SET_GENERIC,
   4833             &bad_member_of,
   4834         );
   4835         assert!(ingest_list_set_event(&exec, &bad_member_of_event).is_err());
   4836 
   4837         let mut bad_plots = farm_list_sets::farm_plots_list_set(
   4838             &farm_d_tag,
   4839             &farm_pubkey,
   4840             vec![plot_d_tag.clone()],
   4841         )
   4842         .expect("plots");
   4843         bad_plots.entries[0].tag = "p".to_string();
   4844         let bad_plots_event =
   4845             list_set_event(952, &farm_pubkey, 222, KIND_LIST_SET_GENERIC, &bad_plots);
   4846         assert!(ingest_list_set_event(&exec, &bad_plots_event).is_err());
   4847 
   4848         let mut bad_members =
   4849             farm_list_sets::farm_members_list_set(&farm_d_tag, vec![fixture_public_key(6)])
   4850                 .expect("members");
   4851         bad_members.entries[0].tag = "a".to_string();
   4852         let bad_members_event =
   4853             list_set_event(953, &farm_pubkey, 223, KIND_LIST_SET_GENERIC, &bad_members);
   4854         assert!(ingest_list_set_event(&exec, &bad_members_event).is_err());
   4855     }
   4856 }
	lib Core libraries for Radroots git clone https://radroots.dev/git/lib.git
	Log \| Files \| Refs \| README