Add SQL mode and make it use the query! macro for type checking

.github/workflows/build.yml

@@ -45,6 +45,8 @@ jobs:
 
       - name: Build release binary
         run: cargo build --release
+        env:
+          SQLX_OFFLINE: true
 
       - name: Upload binary
         uses: actions/upload-artifact@v3

.sqlx/query-ac15bfcd1737751e27388ffddfe8ec47fd7277be74c47a59484e4fc993671d43.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "with timelineItems as (select jsonb_array_elements(raw_files.json -> 'timelineItems') as timelineItem\n                       from raw_files\n                       where date = ANY ($1)),\n     max_last_saved as (select timelineItem ->> 'itemId'                          as itemId,\n                               max((timelineItem ->> 'lastSaved') :: timestamptz) as latest_last_saved\n                        from timelineItems\n                        group by timelineItem ->> 'itemId'),\n     unique_timline_items as (select distinct on (max_last_saved.itemId) *\n                              from max_last_saved\n                                       inner join timelineItems\n                                                  on timelineItems.timelineItem ->> 'itemId' = max_last_saved.itemId\n                                                      and (timelineItems.timelineItem ->> 'lastSaved') :: timestamptz =\n                                                          max_last_saved.latest_last_saved)\ninsert\ninto public.timeline_item (item_id, json, place_id, end_date, last_saved, server_last_updated)\nselect unique_timline_items.itemId :: uuid                                  as item_id,\n       unique_timline_items.timelineItem                                    as json,\n       (unique_timline_items.timelineItem -> 'place' ->> 'placeId') :: uuid as place_id,\n       (unique_timline_items.timelineItem ->> 'endDate') :: timestamptz     as end_date,\n       unique_timline_items.latest_last_saved :: timestamptz                as last_saved,\n       now()                                                                as server_last_updated\nfrom unique_timline_items\non conflict (item_id) do update set json                = excluded.json,\n                                    place_id            = excluded.place_id,\n                                    end_date            = excluded.end_date,\n                                    last_saved          = excluded.last_saved,\n                                    server_last_updated = excluded.server_last_updated\nwhere excluded.last_saved > public.timeline_item.last_saved\nreturning item_id;\n",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "item_id",
+        "type_info": "Uuid"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "TextArray"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "ac15bfcd1737751e27388ffddfe8ec47fd7277be74c47a59484e4fc993671d43"
+}

.sqlx/query-f0402ce4a5c93837f39559bfb3358a7655f178080465f7abd7f7702d42474157.json

@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "with timelineItems as (select jsonb_array_elements(raw_files.json -> 'timelineItems') as timelineItem\n                       from raw_files\n                       where date = ANY ($1)),\n     places as (select distinct on (md5(timelineItem ->> 'place' :: text)) timelineItem -> 'place'                                  as place,\n                                                                           timelineItem -> 'place' ->> 'placeId'                    as placeId,\n                                                                           (timelineItem -> 'place' ->> 'lastSaved') :: timestamptz as lastSaved\n                from timelineItems\n                where timelineItem ->> 'place' is not null),\n     places_with_max_last_saved as (select place -> 'placeId'                          as placeId,\n                                           max((place ->> 'lastSaved') :: timestamptz) as latest_last_saved\n                                    from places\n                                    group by place -> 'placeId'),\n     latest_places as (select places.*\n                       from places_with_max_last_saved\n                                inner join places on places.place -> 'placeId' = places_with_max_last_saved.placeId and\n                                                     places.lastSaved =\n                                                     places_with_max_last_saved.latest_last_saved)\ninsert\ninto public.place (place_id, json, last_saved, server_last_updated)\nselect (placeId :: uuid) as place_id, place as json, lastSaved as last_saved, now() as server_last_updated\nfrom latest_places\non conflict (place_id) do update set json                = excluded.json,\n                                     last_saved          = excluded.last_saved,\n                                     server_last_updated = excluded.server_last_updated\nwhere excluded.last_saved > public.place.last_saved;\n",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "TextArray"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "f0402ce4a5c93837f39559bfb3358a7655f178080465f7abd7f7702d42474157"
+}

Cargo.lock

@@ -100,21 +100,34 @@ dependencies = [
 ]
 
 [[package]]
-name = "arc-ingester"
+name = "anyhow"
-version = "0.1.1"
+version = "1.0.82"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f538837af36e6f6a9be0faa67f9a314f8119e4e4b5867c6ab40ed60360142519"
 dependencies = [
+ "backtrace",
+]
+
+[[package]]
+name = "arc-ingester"
+version = "0.1.2"
+dependencies = [
+ "anyhow",
  "chrono",
  "clap",
  "flate2",
  "futures",
  "itertools",
- "rayon",
  "serde",
  "serde_json",
  "sha256",
  "sqlx",
+ "testcontainers",
+ "testcontainers-modules",
  "thiserror",
  "tokio",
+ "tracing",
+ "tracing-subscriber",
  "uuid",
 ]
 
@@ -205,6 +218,16 @@ dependencies = [
  "generic-array",
 ]
 
+[[package]]
+name = "bollard-stubs"
+version = "1.42.0-rc.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ed59b5c00048f48d7af971b71f800fdf23e858844a6f9e4d32ca72e9399e7864"
+dependencies = [
+ "serde",
+ "serde_with",
+]
+
 [[package]]
 name = "bumpalo"
 version = "3.14.0"
@@ -344,25 +367,6 @@ dependencies = [
  "cfg-if",
 ]
 
-[[package]]
-name = "crossbeam-deque"
-version = "0.8.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "613f8cc01fe9cf1a3eb3d7f488fd2fa8388403e97039e2f73692932e291a770d"
-dependencies = [
- "crossbeam-epoch",
- "crossbeam-utils",
-]
-
-[[package]]
-name = "crossbeam-epoch"
-version = "0.9.18"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5b82ac4a3c2ca9c3460964f020e1402edd5753411d7737aa39c3714ad1b5420e"
-dependencies = [
- "crossbeam-utils",
-]
-
 [[package]]
 name = "crossbeam-queue"
 version = "0.3.11"
@@ -388,6 +392,41 @@ dependencies = [
  "typenum",
 ]
 
+[[package]]
+name = "darling"
+version = "0.13.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a01d95850c592940db9b8194bc39f4bc0e89dee5c4265e4b1807c34a9aba453c"
+dependencies = [
+ "darling_core",
+ "darling_macro",
+]
+
+[[package]]
+name = "darling_core"
+version = "0.13.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "859d65a907b6852c9361e3185c862aae7fafd2887876799fa55f5f99dc40d610"
+dependencies = [
+ "fnv",
+ "ident_case",
+ "proc-macro2",
+ "quote",
+ "strsim",
+ "syn 1.0.109",
+]
+
+[[package]]
+name = "darling_macro"
+version = "0.13.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9c972679f83bdf9c42bd905396b6c3588a843a17f0f16dfcfa3e2c5d57441835"
+dependencies = [
+ "darling_core",
+ "quote",
+ "syn 1.0.109",
+]
+
 [[package]]
 name = "der"
 version = "0.7.8"
@@ -492,6 +531,12 @@ dependencies = [
  "spin 0.9.8",
 ]
 
+[[package]]
+name = "fnv"
+version = "1.0.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1"
+
 [[package]]
 name = "form_urlencoded"
 version = "1.2.1"
@@ -718,6 +763,12 @@ dependencies = [
  "cc",
 ]
 
+[[package]]
+name = "ident_case"
+version = "1.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b9e0384b61958566e926dc50660321d12159025e767c18e043daf26b70104c39"
+
 [[package]]
 name = "idna"
 version = "0.5.0"
@@ -879,6 +930,16 @@ dependencies = [
  "minimal-lexical",
 ]
 
+[[package]]
+name = "nu-ansi-term"
+version = "0.46.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "77a8165726e8236064dbb45459242600304b42a5ea24ee2948e18e023bf7ba84"
+dependencies = [
+ "overload",
+ "winapi",
+]
+
 [[package]]
 name = "num-bigint-dig"
 version = "0.8.4"
@@ -952,6 +1013,12 @@ version = "1.19.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3fdb12b2476b595f9358c5161aa467c2438859caa136dec86c26fdd2efe17b92"
 
+[[package]]
+name = "overload"
+version = "0.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b15813163c1d831bf4a13c3610c05c0d03b39feb07f7e09fa234dac9b15aaf39"
+
 [[package]]
 name = "parking_lot"
 version = "0.12.1"
@@ -1089,26 +1156,6 @@ dependencies = [
  "getrandom",
 ]
 
-[[package]]
-name = "rayon"
-version = "1.8.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fa7237101a77a10773db45d62004a272517633fbcc3df19d96455ede1122e051"
-dependencies = [
- "either",
- "rayon-core",
-]
-
-[[package]]
-name = "rayon-core"
-version = "1.12.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1465873a3dfdaa8ae7cb14b4383657caab0b3e8a0aa9ae8e04b044854c8dfce2"
-dependencies = [
- "crossbeam-deque",
- "crossbeam-utils",
-]
-
 [[package]]
 name = "redox_syscall"
 version = "0.4.1"
@@ -1254,6 +1301,28 @@ dependencies = [
  "serde",
 ]
 
+[[package]]
+name = "serde_with"
+version = "1.14.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "678b5a069e50bf00ecd22d0cd8ddf7c236f68581b03db652061ed5eb13a312ff"
+dependencies = [
+ "serde",
+ "serde_with_macros",
+]
+
+[[package]]
+name = "serde_with_macros"
+version = "1.5.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e182d6ec6f05393cc0e5ed1bf81ad6db3a8feedf8ee515ecdd369809bcce8082"
+dependencies = [
+ "darling",
+ "proc-macro2",
+ "quote",
+ "syn 1.0.109",
+]
+
 [[package]]
 name = "sha1"
 version = "0.10.6"
@@ -1289,6 +1358,15 @@ dependencies = [
  "tokio",
 ]
 
+[[package]]
+name = "sharded-slab"
+version = "0.1.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f40ca3c46823713e0d4209592e8d6e826aa57e928f09752619fc696c499637f6"
+dependencies = [
+ "lazy_static",
+]
+
 [[package]]
 name = "signal-hook-registry"
 version = "1.4.1"
@@ -1634,6 +1712,32 @@ dependencies = [
  "windows-sys 0.52.0",
 ]
 
+[[package]]
+name = "testcontainers"
+version = "0.15.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f83d2931d7f521af5bae989f716c3fa43a6af9af7ec7a5e21b59ae40878cec00"
+dependencies = [
+ "bollard-stubs",
+ "futures",
+ "hex",
+ "hmac",
+ "log",
+ "rand",
+ "serde",
+ "serde_json",
+ "sha2",
+]
+
+[[package]]
+name = "testcontainers-modules"
+version = "0.3.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "204d1c7516bfdc8a01bb85d3e30145e5bbeb2351812e5e8aa6971769109b45b5"
+dependencies = [
+ "testcontainers",
+]
+
 [[package]]
 name = "thiserror"
 version = "1.0.56"
@@ -1654,6 +1758,16 @@ dependencies = [
  "syn 2.0.48",
 ]
 
+[[package]]
+name = "thread_local"
+version = "1.1.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8b9ef9bad013ada3808854ceac7b46812a6465ba368859a37e2100283d2d719c"
+dependencies = [
+ "cfg-if",
+ "once_cell",
+]
+
 [[package]]
 name = "tinyvec"
 version = "1.6.0"
@@ -1740,6 +1854,32 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c06d3da6113f116aaee68e4d601191614c9053067f9ab7f6edbcb161237daa54"
 dependencies = [
  "once_cell",
+ "valuable",
+]
+
+[[package]]
+name = "tracing-log"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ee855f1f400bd0e5c02d150ae5de3840039a3f54b025156404e34c23c03f47c3"
+dependencies = [
+ "log",
+ "once_cell",
+ "tracing-core",
+]
+
+[[package]]
+name = "tracing-subscriber"
+version = "0.3.18"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ad0f048c97dbd9faa9b7df56362b8ebcaa52adb06b498c050d2f4e32f90a7a8b"
+dependencies = [
+ "nu-ansi-term",
+ "sharded-slab",
+ "smallvec",
+ "thread_local",
+ "tracing-core",
+ "tracing-log",
 ]
 
 [[package]]
@@ -1819,6 +1959,12 @@ dependencies = [
  "serde",
 ]
 
+[[package]]
+name = "valuable"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "830b7e5d4d90034032940e4ace0d9a9a057e7a45cd94e6c007832e39edb82f6d"
+
 [[package]]
 name = "vcpkg"
 version = "0.2.15"
@@ -1903,6 +2049,28 @@ version = "1.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "22fc3756b8a9133049b26c7f61ab35416c130e8c09b660f5b3958b446f52cc50"
 
+[[package]]
+name = "winapi"
+version = "0.3.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5c839a674fcd7a98952e593242ea400abe93992746761e38641405d28b00f419"
+dependencies = [
+ "winapi-i686-pc-windows-gnu",
+ "winapi-x86_64-pc-windows-gnu",
+]
+
+[[package]]
+name = "winapi-i686-pc-windows-gnu"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6"
+
+[[package]]
+name = "winapi-x86_64-pc-windows-gnu"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f"
+
 [[package]]
 name = "windows-core"
 version = "0.52.0"

Cargo.toml

@@ -1,9 +1,18 @@
 [package]
 name = "arc-ingester"
-version = "0.1.1"
+version = "0.1.2"
 edition = "2021"
 authors = ["Joshua Coles"]
 license = "MIT OR Apache-2.0"
+default-run = "arc-ingester"
+
+[[bin]]
+name = "arc-ingester"
+path = "src/main.rs"
+
+[[bin]]
+name = "t1"
+path = "src/t1.rs"
 
 [dependencies]
 clap = { version = "4.4.18", features = ["derive", "env"] }
@@ -14,8 +23,12 @@ thiserror = "1"
 sha256 = "1"
 serde = { version = "1.0.196", features = ["derive"] }
 serde_json = "1.0.113"
-rayon = "1.8.1"
 futures = "0.3.30"
 itertools = "0.12.1"
 chrono = { version = "0.4.33", features = ["serde"] }
 uuid = { version = "1.7.0", features = ["serde"] }
+testcontainers = "0.15.0"
+testcontainers-modules = { version = "0.3.6", features = ["postgres"] }
+anyhow = { version = "1.0.82", features = ["backtrace"] }
+tracing = "0.1.40"
+tracing-subscriber = "0.3.18"

build.rs

@@ -2,4 +2,5 @@
 fn main() {
     // trigger recompilation when a new migration is added
     println!("cargo:rerun-if-changed=migrations");
+    println!("cargo:rustc-env=DATABASE_URL=postgres://joshuacoles@localhost/arc_test");
 }

functions/update_places.sql

@@ -0,0 +1,25 @@
+with timelineItems as (select jsonb_array_elements(raw_files.json -> 'timelineItems') as timelineItem
+                       from raw_files
+                       where date = ANY ($1)),
+     places as (select distinct on (md5(timelineItem ->> 'place' :: text)) timelineItem -> 'place'                                  as place,
+                                                                           timelineItem -> 'place' ->> 'placeId'                    as placeId,
+                                                                           (timelineItem -> 'place' ->> 'lastSaved') :: timestamptz as lastSaved
+                from timelineItems
+                where timelineItem ->> 'place' is not null),
+     places_with_max_last_saved as (select place -> 'placeId'                          as placeId,
+                                           max((place ->> 'lastSaved') :: timestamptz) as latest_last_saved
+                                    from places
+                                    group by place -> 'placeId'),
+     latest_places as (select places.*
+                       from places_with_max_last_saved
+                                inner join places on places.place -> 'placeId' = places_with_max_last_saved.placeId and
+                                                     places.lastSaved =
+                                                     places_with_max_last_saved.latest_last_saved)
+insert
+into public.place (place_id, json, last_saved, server_last_updated)
+select (placeId :: uuid) as place_id, place as json, lastSaved as last_saved, now() as server_last_updated
+from latest_places
+on conflict (place_id) do update set json                = excluded.json,
+                                     last_saved          = excluded.last_saved,
+                                     server_last_updated = excluded.server_last_updated
+where excluded.last_saved > public.place.last_saved;

functions/update_timeline_items.sql

@@ -0,0 +1,29 @@
+with timelineItems as (select jsonb_array_elements(raw_files.json -> 'timelineItems') as timelineItem
+                       from raw_files
+                       where date = ANY ($1)),
+     max_last_saved as (select timelineItem ->> 'itemId'                          as itemId,
+                               max((timelineItem ->> 'lastSaved') :: timestamptz) as latest_last_saved
+                        from timelineItems
+                        group by timelineItem ->> 'itemId'),
+     unique_timline_items as (select distinct on (max_last_saved.itemId) *
+                              from max_last_saved
+                                       inner join timelineItems
+                                                  on timelineItems.timelineItem ->> 'itemId' = max_last_saved.itemId
+                                                      and (timelineItems.timelineItem ->> 'lastSaved') :: timestamptz =
+                                                          max_last_saved.latest_last_saved)
+insert
+into public.timeline_item (item_id, json, place_id, end_date, last_saved, server_last_updated)
+select unique_timline_items.itemId :: uuid                                  as item_id,
+       unique_timline_items.timelineItem                                    as json,
+       (unique_timline_items.timelineItem -> 'place' ->> 'placeId') :: uuid as place_id,
+       (unique_timline_items.timelineItem ->> 'endDate') :: timestamptz     as end_date,
+       unique_timline_items.latest_last_saved :: timestamptz                as last_saved,
+       now()                                                                as server_last_updated
+from unique_timline_items
+on conflict (item_id) do update set json                = excluded.json,
+                                    place_id            = excluded.place_id,
+                                    end_date            = excluded.end_date,
+                                    last_saved          = excluded.last_saved,
+                                    server_last_updated = excluded.server_last_updated
+where excluded.last_saved > public.timeline_item.last_saved
+returning item_id;

src/main.rs

@@ -2,13 +2,12 @@ use std::collections::HashMap;
 use std::fs::ReadDir;
 use std::io::Read;
 use std::path::PathBuf;
-use std::sync::Arc;
 use chrono::{DateTime, Utc};
 use clap::Parser;
 use itertools::Itertools;
-use sqlx::FromRow;
+use sqlx::{Executor, FromRow, PgPool, Pool, Postgres};
-use rayon::prelude::*;
 use serde_json::Value;
+use tracing::instrument;
 use uuid::Uuid;
 
 #[derive(Parser, Debug)]
@@ -20,6 +19,9 @@ struct Cli {
     /// psql connection string
     #[arg(long, short, env = "ARC_DB")]
     conn: String,
+
+    #[arg(long)]
+    sql_only: bool,
 }
 
 #[derive(Debug, FromRow)]
@@ -69,88 +71,129 @@ struct Place {
     rest: HashMap<String, Value>,
 }
 
-async fn find_updated(db: &sqlx::PgPool, files: ReadDir) -> Vec<UpdatedFile> {
+fn hash_files(files: ReadDir) -> impl Iterator<Item=(String, PathBuf, String)> {
-    let date_hashes: Vec<DateHash> = sqlx::query_as("SELECT date, sha256 FROM raw_files")
+    files.into_iter()
-        .fetch_all(db)
-        .await
-        .unwrap();
-
-    let date_hashes: Arc<HashMap<String, String>> = Arc::new(date_hashes.into_iter()
-        .map(|dh| (dh.date, dh.sha256))
-        .collect());
-
-    let date_hashes = &date_hashes;
-
-    // Find the files that need to be refreshed, based on the sha256 hash of the file compared to
-    // the hash stored in the database.
-    let need_refresh = files.into_iter()
         .map(|f| f.unwrap().path())
-        .filter_map(|path| {
+        .filter(|p| p.extension().and_then(|ext| ext.to_str()).filter(|p| *p == "gz").is_some())
-            // Extract the date from the file name
+        .map(|path| {
             let date = {
                 let file_name = path.file_name().unwrap().to_str().unwrap();
                 let i = file_name.find('.').unwrap();
                 &file_name[..i]
             };
 
-            let current_hash = sha256::try_digest(&path).unwrap();
+            let hash = sha256::try_digest(&path).unwrap();
-            let existing_hash = date_hashes.get(date);
+            (date.to_string(), path, hash)
-
+        })
-            if let Some(existing_hash) = existing_hash {
-                if current_hash == *existing_hash {
-                    return None;
 }
+
+#[instrument(skip(db, files))]
+async fn find_updated(db: &PgPool, files: ReadDir) -> Vec<UpdatedFile> {
+    let date_hashes: Vec<DateHash> = sqlx::query_as("SELECT date, sha256 FROM raw_files")
+        .fetch_all(db)
+        .await
+        .expect("Failed to fetch date hashes from database");
+
+    let date_hashes: HashMap<String, String> = date_hashes.into_iter()
+        .map(|dh| (dh.date, dh.sha256))
+        .collect();
+
+    let new_hashes = hash_files(files);
+
+    new_hashes.filter_map(|(date, path, new_hash)| {
+        let span = tracing::span!(tracing::Level::DEBUG, "considering_file", path = ?path, date = ?date);
+        let _enter = span.enter();
+
+        tracing::debug!("Considering file for updates {path:?} (read as date: {date})");
+        if let Some(existing_hash) = date_hashes.get(&date) {
+            if new_hash == *existing_hash {
+                tracing::debug!("Found and matched to to existing hash");
+                return None;
+            } else {
+                tracing::debug!("Found existing hash but file does not match");
+            }
+        } else {
+            tracing::debug!("No existing hash found");
         }
 
         let bytes = std::fs::read(&path).unwrap();
-
-            return Some((date.to_string(), current_hash, bytes));
-        })
-        .collect_vec();
-
-    let decompressed = need_refresh.par_iter().map(|(date, new_hash, bytes)| {
         let mut decoder = flate2::bufread::GzDecoder::new(&bytes[..]);
         let mut string = String::new();
-        decoder.read_to_string(&mut string).unwrap();
+        match decoder.read_to_string(&mut string) {
+            Err(err) => {
+                eprintln!("Failed to parse file {path:?}");
+                eprintln!("Error {err:?}");
+                panic!("Error")
+            }
+            _ => {}
+        }
 
-        UpdatedFile {
+        Some(UpdatedFile {
             date: date.clone(),
             sha256: new_hash.clone(),
             json: serde_json::from_str(&string).unwrap(),
-        }
+        })
-    }).collect::<Vec<_>>();
+    }).collect_vec()
-
-    decompressed
 }
 
 #[tokio::main]
-async fn main() {
+async fn main() -> anyhow::Result<()> {
+    tracing_subscriber::fmt::init();
+
     let cli = Cli::parse();
     let daily_exports = cli.root.join("Export/JSON/Daily");
-    let files = std::fs::read_dir(daily_exports).unwrap();
+    let files = std::fs::read_dir(daily_exports)
+        .expect("Failed to access daily exports directory");
 
     let db = sqlx::PgPool::connect(&cli.conn)
         .await
-        .unwrap();
+        .expect("Failed to connect to postgres database");
 
     sqlx::migrate!()
         .run(&db)
         .await
-        .unwrap();
+        .expect("Failed to migrate postgres database");
 
-    let need_refresh = find_updated(&db, files).await;
+    let need_refresh = find_updated(&db, files)
+        .await;
 
-    // Refresh the database with the new files
+    upload_files(&db, &need_refresh)
-    for updated_file in &need_refresh {
+        .await?;
-        sqlx::query("INSERT INTO raw_files (date, sha256, json) VALUES ($1, $2, $3 :: jsonb) ON CONFLICT (date) DO UPDATE SET sha256 = $2, json = $3 :: jsonb")
+
-            .bind(&updated_file.date)
+    if cli.sql_only {
-            .bind(&updated_file.sha256)
+        update_data_sql(&db, need_refresh)
-            .bind(&serde_json::to_value(&updated_file.json).unwrap())
+            .await?;
-            .execute(&db)
+    } else {
-            .await
+        update_data(&db, need_refresh)
-            .unwrap();
+            .await?;
     }
 
+    Ok(())
+}
+
+#[instrument(skip(db, need_refresh))]
+async fn update_data_sql(db: &PgPool, need_refresh: Vec<UpdatedFile>) -> anyhow::Result<()> {
+    let vec = need_refresh.iter()
+        .map(|d| d.date.clone())
+        .collect_vec();
+
+    let result = sqlx::query_file!("functions/update_places.sql", &vec)
+        .execute(db)
+        .await?;
+
+    tracing::info!("Updated {} places", result.rows_affected());
+
+    let updated = sqlx::query_file!("functions/update_timeline_items.sql", &vec)
+        .fetch_all(db)
+        .await?;
+
+    tracing::info!("Updated {} timeline items", updated.len());
+
+    Ok(())
+}
+
+#[instrument(skip(db, need_refresh))]
+async fn update_data(db: &Pool<Postgres>, need_refresh: Vec<UpdatedFile>) -> anyhow::Result<()> {
     // Take all the changed files' timeline items, and group them by item_id, then take the latest one.
     // If we are needing to update the database it will be with this one.
     let possibly_timeline_items = need_refresh.into_iter()
@@ -167,9 +210,8 @@ async fn main() {
 
     let existing_last_saved_at_map = sqlx::query_as("SELECT item_id, end_date, last_saved FROM timeline_item where item_id = ANY($1)")
         .bind(&possibly_timeline_item_ids)
-        .fetch_all(&db)
+        .fetch_all(db)
-        .await
+        .await?
-        .unwrap()
         .into_iter()
         .map(|row: TimelineItemUpdatedCheckRow| (row.item_id, row.last_saved))
         .collect::<HashMap<_, _>>();
@@ -190,18 +232,37 @@ async fn main() {
         if let Some(place) = &updated_timeline_item.place {
             sqlx::query("INSERT INTO place (place_id, json, last_saved, server_last_updated) VALUES ($1, $2 :: jsonb, $3, now()) ON CONFLICT (place_id) DO UPDATE SET json = $2 :: jsonb, last_saved = $3, server_last_updated = now()")
                 .bind(&place.place_id)
-                .bind(&serde_json::to_value(&place).unwrap())
+                .bind(&serde_json::to_value(&place)?)
                 .bind(&place.last_saved)
-                .execute(&db).await.unwrap();
+                .execute(db)
+                .await?;
         }
 
         // Then we can insert/update the timeline item.
         sqlx::query("INSERT INTO timeline_item (item_id, json, place_id, end_date, last_saved, server_last_updated) VALUES ($1, $2 :: jsonb, $3, $4, $5, now()) ON CONFLICT (item_id) DO UPDATE SET json = $2 :: jsonb, place_id = $3, end_date = $4, last_saved = $5, server_last_updated = now()")
             .bind(&updated_timeline_item.item_id)
-            .bind(&serde_json::to_value(&updated_timeline_item).unwrap())
+            .bind(&serde_json::to_value(&updated_timeline_item)?)
             .bind(&updated_timeline_item.place.map(|place| place.place_id))
             .bind(&updated_timeline_item.end_date)
             .bind(&updated_timeline_item.last_saved)
-            .execute(&db).await.unwrap();
+            .execute(db)
+            .await?;
     }
+
+    Ok(())
+}
+
+#[tracing::instrument(skip(db, need_refresh), err)]
+async fn upload_files(db: &PgPool, need_refresh: &[UpdatedFile]) -> anyhow::Result<()> {
+    // Refresh the database with the new files
+    for updated_file in need_refresh {
+        sqlx::query("INSERT INTO raw_files (date, sha256, json) VALUES ($1, $2, $3 :: jsonb) ON CONFLICT (date) DO UPDATE SET sha256 = excluded.sha256, json = excluded.json")
+            .bind(&updated_file.date)
+            .bind(&updated_file.sha256)
+            .bind(&serde_json::to_value(&updated_file.json)?)
+            .execute(db)
+            .await?;
+    }
+
+    Ok(())
 }

src/t1.rs

@@ -0,0 +1,16 @@
+use sqlx::{Connection, Executor};
+use testcontainers::core::env::Os;
+
+#[tokio::main]
+async fn main() {
+    let tc = testcontainers::clients::Cli::new::<Os>();
+    let pg_spec = testcontainers_modules::postgres::Postgres::default();
+    let pg_container = tc.run(pg_spec);
+    pg_container.start();
+    println!("postgres running");
+    let mut pg = sqlx::postgres::PgConnection::connect(&format!("postgres://postgres:postgres@localhost:{}/postgres", pg_container.get_host_port_ipv4(5432)),)
+        .await
+        .unwrap();
+
+    dbg!(pg.execute("select 1").await.unwrap());
+}