fix: Tangled dispatch correctness + cleanup dead code - Move sh.tangled.repo/pipeline/refUpdate/issue/pull from simple dispatch to special-case arms so they get node URL lookup, breaking change counting, SSE events, and algebraic check extraction - Add shared util.rs with consolidated camel_to_snake, nsid_to_pascal, find_record_body, is_field_required (was triplicated) - Remove dead auto_camel_to_snake_renames from db_projection.rs

8080 ) -> anyhow::Result<()> {
8181     match collection {
8282         // ─── Repo (node URL lookup) ─────────────────────────────────
83-        "dev.cospan.repo" => {
83+        "dev.cospan.repo" | "sh.tangled.repo" => {
8484             let node_uri = rec.get("node").and_then(|v| v.as_str()).unwrap_or("");
8585             let node_did = extract_did_from_at_uri(node_uri);
8686 

104104         }
105105 
106106         // ─── Ref Update (breaking change count + SSE) ───────────────
107-        "dev.cospan.vcs.refUpdate" => {
107+        "dev.cospan.vcs.refUpdate" | "sh.tangled.git.refUpdate" => {
108108             let breaking_changes = rec
109109                 .get("breakingChanges")
110110                 .and_then(|v| v.as_array())

129129         }
130130 
131131         // ─── Issue (SSE event on create) ────────────────────────────
132-        "dev.cospan.repo.issue" => {
132+        "dev.cospan.repo.issue" | "sh.tangled.repo.issue" => {
133133             let mut row: db::issue::IssueRow =
134134                 serde_json::from_value(transform_record(state, collection, rec))?;
135135             row.did = did.to_string();

226226         }
227227 
228228         // ─── Pull Request (SSE event on create) ─────────────────────
229-        "dev.cospan.repo.pull" => {
229+        "dev.cospan.repo.pull" | "sh.tangled.repo.pull" => {
230230             let mut row: db::pull::PullRow =
231231                 serde_json::from_value(transform_record(state, collection, rec))?;
232232             row.did = did.to_string();

336336         }
337337 
338338         // ─── Pipeline (algebraicChecks extraction) ──────────────────
339-        "dev.cospan.pipeline" => {
339+        "dev.cospan.pipeline" | "sh.tangled.pipeline" => {
340340             let checks = rec.get("algebraicChecks");
341341 
342342             let mut row: db::pipeline::PipelineRow =

8080         "sh.tangled.feed.reaction" => simple_upsert!(reaction, ReactionRow),
8181         "sh.tangled.repo.collaborator" => simple_upsert!(collaborator, CollaboratorRow),
8282         "sh.tangled.knot.member" => simple_upsert!(org_member, OrgMemberRow),
83-        "sh.tangled.repo" => simple_upsert!(repo, RepoRow),
84-        "sh.tangled.pipeline" => simple_upsert!(pipeline, PipelineRow),
85-        "sh.tangled.git.refUpdate" => simple_upsert!(ref_update, RefUpdateRow, no_did),
86-        "sh.tangled.repo.issue" => simple_upsert!(issue, IssueRow),
87-        "sh.tangled.repo.pull" => simple_upsert!(pull, PullRow),
88-
83+        // NOTE: sh.tangled.repo, pipeline, refUpdate, issue, pull have side effects
84+        // (node URL lookup, breaking change count, SSE events) and are handled
85+        // in consumer.rs dispatch_special_upsert, not here.
8986         _ => Ok(false),
9087     }
9188 }

324324     }
325325 }
326326 
327-/// Auto-generate camelCase → snake_case renames for all Lexicon fields.
328-/// This ensures `to_json()` output has snake_case keys matching Row struct field names.
329-pub fn auto_camel_to_snake_renames(
330-    schema: &panproto_schema::Schema,
331-    nsid: &str,
332-) -> Vec<FieldTransform> {
333-    let body_vertex = record_body_vertex(nsid);
334-    let props = panproto_protocols::emit::children_by_edge(schema, &body_vertex, "prop");
335-    let mut renames = Vec::new();
336-    for (edge, _) in &props {
337-        if let Some(name) = edge.name.as_ref() {
338-            let camel = name.as_str();
339-            let snake = camel_to_snake(camel);
340-            if camel != snake {
341-                renames.push(rename_field(camel, &snake));
342-            }
343-        }
344-    }
345-    renames
346-}
347-
348-fn camel_to_snake(s: &str) -> String {
349-    let mut r = String::with_capacity(s.len() + 4);
350-    for (i, c) in s.chars().enumerate() {
351-        if c.is_uppercase() && i > 0 {
352-            r.push('_');
353-        }
354-        r.push(c.to_lowercase().next().unwrap_or(c));
355-    }
356-    r
357-}
358-
359327 /// Get the record body vertex ID for a given NSID.
360328 /// ATProto Lexicon schemas have the body under `{nsid}.record`.
361329 fn record_body_vertex(nsid: &str) -> String {

136136         }
137137 
138138         if let Some(ovr) = type_override {
139+            let is_optional = ovr.rust_type.starts_with("Option<");
139140             cols.push(Column {
140141                 name: snake,
141142                 camel_name: field_name.into(),
142143                 rust_type: ovr.rust_type.into(),
143144                 sql_type: ovr.sql_type.into(),
144-                optional: false,
145+                optional: is_optional,
145146                 is_counter: false,
146147             });
147148         } else {

55 pub mod morphism;
66 pub mod record_config;
77 pub mod tangled_interop;
8+pub mod util;