refactor(jetstream,knot): restructure jetstream types

Signed-off-by: tjh <did:plc:65gha4t3avpfpzmvpbwovss7>

crates/jetstream/src/types.rs

···
 #[derive(Debug, Deserialize)]
 #[serde(from = "ShadowEventPayload<T>")]
 pub enum EventPayload<'a, T> {
-    #[serde(borrow)]
-    Commit(Commit<'a, T>),
+    CommitCreate(#[serde(borrow)] CommitMeta<'a>, CommitData<'a, T>),
+    CommitUpdate(#[serde(borrow)] CommitMeta<'a>, CommitData<'a, T>),
+    CommitDelete(#[serde(borrow)] CommitMeta<'a>),
     Account(Account<'a>),
     Identity(Identity<'a>),
 }
···
 impl<'a, T> From<ShadowEventPayload<'a, T>> for EventPayload<'a, T> {
     fn from(value: ShadowEventPayload<'a, T>) -> Self {
         match value {
-            ShadowEventPayload::Commit { commit } => EventPayload::Commit(commit),
+            ShadowEventPayload::Commit {
+                commit:
+                    Commit::Create {
+                        collection,
+                        rkey,
+                        rev,
+                        cid,
+                        record,
+                    },
+            } => EventPayload::CommitCreate(
+                CommitMeta {
+                    collection,
+                    rkey,
+                    rev,
+                },
+                CommitData { cid, record },
+            ),
+            ShadowEventPayload::Commit {
+                commit:
+                    Commit::Update {
+                        collection,
+                        rkey,
+                        rev,
+                        cid,
+                        record,
+                    },
+            } => EventPayload::CommitCreate(
+                CommitMeta {
+                    collection,
+                    rkey,
+                    rev,
+                },
+                CommitData { cid, record },
+            ),
+            ShadowEventPayload::Commit {
+                commit:
+                    Commit::Delete {
+                        collection,
+                        rkey,
+                        rev,
+                    },
+            } => EventPayload::CommitDelete(CommitMeta {
+                collection,
+                rkey,
+                rev,
+            }),
             ShadowEventPayload::Account { account } => EventPayload::Account(account),
             ShadowEventPayload::Identity { identity } => EventPayload::Identity(identity),
         }
···
 #[serde(rename_all = "snake_case")]
 #[serde(tag = "operation")]
 pub enum Commit<'a, T> {
-    Create(#[serde(borrow)] CommitData<'a, T>),
-    Update(#[serde(borrow)] CommitData<'a, T>),
+    Create {
+        #[serde(borrow)]
+        collection: &'a Nsid,
+        rkey: &'a str,
+        rev: &'a str,
+        cid: &'a str,
+        record: T,
+    },
+    Update {
+        #[serde(borrow)]
+        collection: &'a Nsid,
+        rkey: &'a str,
+        rev: &'a str,
+        cid: &'a str,
+        record: T,
+    },
     Delete {
         #[serde(borrow)]
         collection: &'a Nsid,
···
     },
 }
 
-#[derive(Debug, Deserialize)]
-pub struct CommitData<'a, T> {
+#[derive(Debug, Hash, PartialEq, Eq, Deserialize)]
+pub struct CommitMeta<'a> {
     #[serde(borrow)]
     pub collection: &'a Nsid,
     pub rkey: &'a str,
     pub rev: &'a str,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct CommitData<'a, T> {
     pub cid: &'a str,
     pub record: T,
 }

crates/knot/src/main.rs

···
+mod cli;
+
 use axum::{
     Router,
     extract::{Query, WebSocketUpgrade},
···
 use tracing::{Span, field::Empty};
 use tracing_subscriber::{EnvFilter, layer::SubscriberExt as _, util::SubscriberInitExt as _};
 use url::Url;
-
-mod cli;
 
 fn main() {
     let stderr_layer = tracing_subscriber::fmt::layer().with_writer(std::io::stderr);
···
     {
         let knot = knot.clone();
         tokio::task::spawn(async move {
-            use jetstream::types::{Commit, EventPayload};
+            use jetstream::types::EventPayload;
 
             while let Some(event) = jetstream_rx.recv_async().await {
-                let Ok(event) = event.deserialize() else {
-                    tracing::error!("failed to deserialize event");
-                    continue;
+                let event = match event.deserialize() {
+                    Ok(event) => event,
+                    Err(error) => {
+                        tracing::error!(?error, "failed to deserialize event");
+                        continue;
+                    }
                 };
 
-                tracing::debug!(?event);
+                tracing::trace!(?event);
 
-                if let EventPayload::Commit(Commit::Create(create)) = event.payload {
-                    tracing::info!("{create:?}");
-                    if let Lexicon::PublicKey(public_key) = create.record {
-                        tracing::info!(?public_key);
-                        if let Ok(pk) = oauth::public_key::PublicKey::from_openssh(&public_key.key)
-                        {
-                            tracing::info!("adding public key to cache");
-                            knot.insert_public_keys(event.did.to_owned(), [pk].into_iter())
-                                .await;
-                        }
+                match event.payload {
+                    EventPayload::CommitCreate(meta, commit)
+                    | EventPayload::CommitUpdate(meta, commit) => {
+                        tracing::info!(?meta, ?commit, "jetstream create/update");
+                    }
+                    EventPayload::CommitDelete(meta) => {
+                        tracing::info!(?meta, "jetstream delete");
+                    }
+                    EventPayload::Account(_) | EventPayload::Identity(_) => {
+                        knot.resolver().invalidate_did(event.did).await;
                     }
                 }
             }