diff --git a/.env.template b/.env.template
index dcd3a22..5d40cdd 100644
--- a/.env.template
+++ b/.env.template
@@ -3,6 +3,6 @@ USER_DID=did:plc:4zht3z4caxwrw3dlsybodywc # did
 DATABASE_URL=postgres://admin@localhost:5432/meview # postgres uri
 
 # optional 
-USER_PDS_URL=katproto.girlonthemoon.xyz # if ommited, will resolve from did
-USER_EXPORT_URL=katproto.girlonthemoon.xyz # if ommited, will copy USER_PDS_URL
-USER_SUBSCRIBE_URL=katproto.girlonthemoon.xyz # if ommited, will copy USER_PDS_URL
\ No newline at end of file
+USER_PDS_URL=abnormal.zip # if ommited, will resolve from did
+USER_EXPORT_URL=abnormal.zip # if ommited, will copy USER_PDS_URL
+USER_SUBSCRIBE_URL=abnormal.zip # if ommited, will copy USER_PDS_URL
\ No newline at end of file
diff --git a/src/ingest/ingest.rs b/src/ingest/ingest.rs
index 247b1d2..cf67c8e 100644
--- a/src/ingest/ingest.rs
+++ b/src/ingest/ingest.rs
@@ -64,18 +64,33 @@ impl Ingest for Commit<'_> {
                 return;
             };
             match op.action.clone().as_str() {
-                "create" => {
+                "create" | "update" => {
                     let Some(cid) = op.cid.map(|x| x.0.to_string()) else {
-                        eprintln!("Missing cid for create {}/{}", collection, rkey);
+                        eprintln!(
+                            "Missing cid for {} {}/{}",
+                            op.action.clone().as_str(),
+                            collection,
+                            rkey
+                        );
                         return;
                     };
                     let Some(val) = val else {
-                        eprintln!("Missing value for create {}/{}/{}", collection, rkey, cid);
+                        eprintln!(
+                            "Missing value for {} {}/{}/{}",
+                            op.action.clone().as_str(),
+                            collection,
+                            rkey,
+                            cid
+                        );
                         return;
                     };
                     if let Err(err) = query!(
                         "INSERT INTO records (collection, rkey, cid, record) 
-						VALUES ($1, $2, $3, $4)",
+						VALUES ($1, $2, $3, $4)
+						ON CONFLICT (collection, rkey)
+						DO UPDATE SET
+							cid = EXCLUDED.cid,
+							record = EXCLUDED.record;",
                         collection,
                         rkey,
                         cid,
@@ -84,40 +99,22 @@ impl Ingest for Commit<'_> {
                     .execute(&*conn)
                     .await
                     {
-                        eprintln!("Error creating {}/{}/{}\n{}", collection, rkey, cid, err);
+                        eprintln!(
+                            "Error applying {} to {}/{}/{}\n{}",
+                            op.action.clone().as_str(),
+                            collection,
+                            rkey,
+                            cid,
+                            err
+                        );
                     } else {
-                        println!("wrote {}/{}/{} successfully.", collection, rkey, cid);
-                    };
-                }
-                "update" => {
-                    let Some(cid) = op.cid.map(|x| x.0.to_string()) else {
-                        eprintln!("Missing cid for update {}/{}", collection, rkey);
-                        return;
-                    };
-                    let Some(val) = val else {
-                        eprintln!("Missing value for update {}/{}/{}", collection, rkey, cid);
-                        return;
-                    };
-                    if let Err(err) = query!(
-                        "UPDATE records SET 
-							collection = $1, 
-							rkey = $2, 
-							cid = $3, 
-							record = $4
-						WHERE 
-							collection = $1 
-							and rkey = $2",
-                        collection,
-                        rkey,
-                        cid,
-                        val
-                    )
-                    .execute(&*conn)
-                    .await
-                    {
-                        eprintln!("Error updating {}/{}/{}\n{}", collection, rkey, cid, err);
-                    } else {
-                        println!("updated {}/{}/{} successfully.", collection, rkey, cid);
+                        println!(
+                            "{} {}/{}/{}",
+                            op.action.clone().as_str(),
+                            collection,
+                            rkey,
+                            cid
+                        );
                     };
                 }
                 "delete" => {
@@ -133,11 +130,16 @@ impl Ingest for Commit<'_> {
                     {
                         eprintln!("Error deleting {}/{}\n{}", collection, rkey, err);
                     } else {
-                        println!("deleted {}/{} successfully.", collection, rkey);
+                        println!("delete {}/{}", collection, rkey);
                     };
                 }
                 _ => {
-                    println!("missing #{} {:#?} {:#?}", op.action.as_str(), op, val)
+                    println!(
+                        "unknown action {} for {:#?} {:#?}",
+                        op.action.as_str(),
+                        op,
+                        val
+                    )
                 }
             }
         }))
diff --git a/src/ingest/queue.rs b/src/ingest/queue.rs
index cba95b4..15c91fe 100644
--- a/src/ingest/queue.rs
+++ b/src/ingest/queue.rs
@@ -1,3 +1,5 @@
+use std::thread;
+use std::time::Duration;
 use std::{collections::VecDeque, sync::Arc};
 
 use futures_util::stream::StreamExt;
@@ -37,6 +39,36 @@ pub async fn queue() -> (
                     Ok(val) => val,
                     Err(err) => {
                         eprintln!("Warning: Websocket error: {} ({:?})", err, err.source());
+                        if &jacquard::StreamErrorKind::Closed == err.kind() {
+                            let stream = client.subscribe(&SubscribeRepos::new().build()).await;
+                            // if it reconnected successfully, just continue
+                            let new_messages = match stream {
+                                Ok(val) => val.into_stream().1,
+                                // if it failed, try reconnect 10 times, waiting a second between each attempt
+                                Err(_) => {
+                                    let mut new_messages = None;
+                                    for _ in 0..10 {
+                                        let Ok(stream) =
+                                            client.subscribe(&SubscribeRepos::new().build()).await
+                                        else {
+                                            // wait a second
+                                            thread::sleep(Duration::from_secs(1));
+                                            continue;
+                                        };
+                                        new_messages = Some(stream.into_stream().1)
+                                    }
+
+                                    if let Some(new_messages) = new_messages {
+                                        new_messages
+                                    } else {
+                                        // could not reconnect so just die lmao
+                                        panic!("Could not reconnect to client. Fatal");
+                                    }
+                                }
+                            };
+                            // for some reason new_messages doesnt need to be mut ?
+                            messages = new_messages;
+                        }
                         continue;
                     }
                 };