Compare commits
10
Commits
99523430c6
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
 joshuacoles
|
e73123b7f9 | ||
|
joshuacoles
|
ef15a630d6 | ||
|
joshuacoles
|
56829cfa58 | ||
|
joshuacoles
|
56a60d969d | ||
|
joshuacoles
|
e0806a2c94 | ||
|
joshuacoles
|
9985485e83 | ||
|
joshuacoles
|
261baa6b5b | ||
|
joshuacoles
|
8cf9d86f32 | ||
|
joshuacoles
|
48f4614b1c | ||
|
joshuacoles
|
bcfa0c2d89 |
@@ -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
|
||||
|
||||
+22
@@ -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"
|
||||
}
|
||||
+14
@@ -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"
|
||||
}
|
||||
Generated
+210
-42
@@ -100,21 +100,34 @@ dependencies = [
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "arc-ingester"
|
||||
version = "0.1.1"
|
||||
name = "anyhow"
|
||||
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"
|
||||
|
||||
+15
-2
@@ -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"
|
||||
|
||||
@@ -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");
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
@@ -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;
|
||||
+122
-61
@@ -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 rayon::prelude::*;
|
||||
use sqlx::{Executor, FromRow, PgPool, Pool, Postgres};
|
||||
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> {
|
||||
let date_hashes: Vec<DateHash> = sqlx::query_as("SELECT date, sha256 FROM raw_files")
|
||||
.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()
|
||||
fn hash_files(files: ReadDir) -> impl Iterator<Item=(String, PathBuf, String)> {
|
||||
files.into_iter()
|
||||
.map(|f| f.unwrap().path())
|
||||
.filter_map(|path| {
|
||||
// Extract the date from the file name
|
||||
.filter(|p| p.extension().and_then(|ext| ext.to_str()).filter(|p| *p == "gz").is_some())
|
||||
.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 existing_hash = date_hashes.get(date);
|
||||
|
||||
if let Some(existing_hash) = existing_hash {
|
||||
if current_hash == *existing_hash {
|
||||
return None;
|
||||
}
|
||||
}
|
||||
|
||||
let bytes = std::fs::read(&path).unwrap();
|
||||
|
||||
return Some((date.to_string(), current_hash, bytes));
|
||||
let hash = sha256::try_digest(&path).unwrap();
|
||||
(date.to_string(), path, hash)
|
||||
})
|
||||
.collect_vec();
|
||||
}
|
||||
|
||||
let decompressed = need_refresh.par_iter().map(|(date, new_hash, bytes)| {
|
||||
#[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();
|
||||
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<_>>();
|
||||
|
||||
decompressed
|
||||
})
|
||||
}).collect_vec()
|
||||
}
|
||||
|
||||
#[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
|
||||
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 = $2, json = $3 :: jsonb")
|
||||
.bind(&updated_file.date)
|
||||
.bind(&updated_file.sha256)
|
||||
.bind(&serde_json::to_value(&updated_file.json).unwrap())
|
||||
.execute(&db)
|
||||
.await
|
||||
.unwrap();
|
||||
upload_files(&db, &need_refresh)
|
||||
.await?;
|
||||
|
||||
if cli.sql_only {
|
||||
update_data_sql(&db, need_refresh)
|
||||
.await?;
|
||||
} else {
|
||||
update_data(&db, need_refresh)
|
||||
.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)
|
||||
.await
|
||||
.unwrap()
|
||||
.fetch_all(db)
|
||||
.await?
|
||||
.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(())
|
||||
}
|
||||
|
||||
@@ -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());
|
||||
}
|
||||
Reference in New Issue
Block a user