Update the other queries to consider source column

crates/corrosion/src/command/consul/sync.rs

@@ -172,6 +172,7 @@ async fn setup(corrosion: &CorrosionClient) -> eyre::Result<()> {
         ("meta", vec![ColumnType::Text, ColumnType::Blob]),
         ("port", vec![ColumnType::Integer]),
         ("address", vec![ColumnType::Text]),
+        ("source", vec![ColumnType::Text]),
         ("updated_at", vec![ColumnType::Integer]),
     ];
 
@@ -203,6 +204,7 @@ async fn setup(corrosion: &CorrosionClient) -> eyre::Result<()> {
         ("name", vec![ColumnType::Text]),
         ("status", vec![ColumnType::Text]),
         ("output", vec![ColumnType::Text]),
+        ("source", vec![ColumnType::Text]),
         ("updated_at", vec![ColumnType::Integer]),
     ];
 
@@ -412,7 +414,8 @@ fn append_upsert_service_statements(
         meta = excluded.meta,
         port = excluded.port,
         address = excluded.address,
-        updated_at = excluded.updated_at;"
+        updated_at = excluded.updated_at
+    WHERE source IS NULL;"
             .into(),
         vec![
             node.into(),
@@ -457,7 +460,8 @@ fn append_upsert_check_statements(
         name = excluded.name,
         status = excluded.status,
         output = excluded.output,
-        updated_at = excluded.updated_at;"
+        updated_at = excluded.updated_at
+    WHERE source IS NULL;"
                                           .into(),vec![
                                               node.into(),
                                               check.id.into(),
@@ -651,7 +655,7 @@ async fn execute(
                     vec![id.clone().into()],
                 ));
                 statements.push(Statement::WithParams(
-                    "DELETE FROM consul_services WHERE node = ? AND id = ?;".into(),
+                    "DELETE FROM consul_services WHERE node = ? AND id = ? AND source IS NULL;".into(),
                     vec![node.into(), id.into()],
                 ));
             }
@@ -677,7 +681,7 @@ async fn execute(
                     vec![id.clone().into()],
                 ));
                 statements.push(Statement::WithParams(
-                    "DELETE FROM consul_checks WHERE node = ? AND id = ?;".into(),
+                    "DELETE FROM consul_checks WHERE node = ? AND id = ? AND source IS NULL;".into(),
                     vec![node.into(), id.into()],
                 ));
             }