Adjust ingestor

crates/ott-ingest/Cargo.toml

···
 tokio-stream = "0.1.17"
 tracing = "0.1.41"
 tracing-subscriber = { version = "0.3.20", features = ["env-filter"] }
-typify = "0.4.3"
 
 [build-dependencies]
 anyhow = "1.0.100"

crates/ott-ingest/src/at_types.rs

···
-use typify::import_types;
-
-import_types!(schema = "schemas/like.json");

crates/ott-ingest/src/lib.rs

···
-pub mod at_types;
 pub mod tei_client;

crates/ott-ingest/src/main.rs

···
+use serde::Deserialize;
 use tokio::{
     select,
     time::{Duration, sleep},
 };
 use tokio_stream::StreamExt;
 use tracing::{debug, error, info, warn};
-use tracing_subscriber::{EnvFilter, fmt};
+use tracing_subscriber::{EnvFilter, field::display, fmt};
 
 use fluvio::{
     Fluvio, Offset,
···
     ops::compute::{CompResult, Op},
     sync::Cache,
 };
-use ott_ingest::at_types;
 use ott_ingest::tei_client;
 
 const LIKES_TOPIC: &str = "raw-likes";
 const POSTS_TOPIC: &str = "raw-posts";
 const PARTITION_NUM: u32 = 0;
 
-#[derive(Clone)]
-struct BskyPost {
-    likes: u32,
+#[derive(Debug, Deserialize, Clone)]
+struct Post {
+    did: String,
     uri: String,
+    #[serde(default)]
+    count: u32,
+}
+
+#[derive(Debug, Deserialize, Clone)]
+struct Like {
     did: String,
-    text: String,
+    #[serde(rename = "commit.record.subject.uri")]
+    uri: String,
 }
-
-impl BskyPost {}
 
 #[tokio::main]
 async fn main() {
···
         .with_env_filter(EnvFilter::from_default_env())
         .init();
 
-    let posts_cache: Cache<String, u32> = Cache::builder()
+    let posts_cache: Cache<String, Post> = Cache::builder()
         .time_to_live(Duration::from_secs(60 * 60))
         .build();
 
···
         let pcc = posts_cache.clone();
         let lcc = posts_cache.clone();
         select! {
-            result = like_stream.next() => {
-                match result {
-                    Some(Ok(record)) => {
-                        record.value();
-                        pcc.entry(String::from("hej"))
-                            .and_compute_with(|maybe_entry| {
-                                if let Some(entry) = maybe_entry {
-                                    let counter = entry.into_value();
-                                    if counter < 2 {
-                                        Op::Put(counter.saturating_add(1)) // Update
-                                    } else {
-                                        Op::Remove
-                                    }
-                                } else {
-                                    Op::Put(1) // Insert
-                                }
-                        });
-                    }
-                    Some(Err(e)) => {todo!()}
-                    None => {todo!()}
-                }
-            }
-
-            Some(Ok(result)) = posts_stream.next() => {
-                lcc.entry(String::from("hej"))
+            Some(Ok(record)) = posts_stream.next() => {
+                let post: Post = serde_json::from_slice(record.value()).unwrap();
+                pcc.entry(post.uri.clone())
+                    .and_compute_with(|maybe_entry| {
+                        if maybe_entry.is_some() {
+                            Op::Nop
+                        } else {
+                            Op::Put(post) // Insert
+                        }
+                });
+            },
+            Some(Ok(record)) = like_stream.next() => {
+                let like: Like = serde_json::from_slice(record.value()).unwrap();
+                lcc.entry(like.uri)
                     .and_compute_with(|maybe_entry| {
                         if let Some(entry) = maybe_entry {
-                            let counter = entry.into_value();
-                            if counter < 2 {
-                                Op::Put(counter.saturating_add(1)) // Update
+                            let mut post = entry.into_value();
+                            if post.count < 2 {
+                                post.count +=1;
+                                warn!("Incread counter for {:#?}", post);
+                                Op::Put(post)
                             } else {
                                 Op::Remove
                             }