Swap to native R2 messages over webhook

.forgejo/workflows/publish-posts.yml

@@ -26,8 +26,3 @@ jobs:
           AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_ACCESS_KEY_SECRET }}
           AWS_DEFAULT_REGION: 'weur'
           AWS_ENDPOINT: ${{ secrets.AWS_ENDPOINT }}
-      - name: Reindex posts
-        run: |
-          curl -H "Authorization: $GABRIELSIMMER_COM_KEY" https://gabrielsimmer.com/api/reindex
-        env:
-          GABRIELSIMMER_COM_KEY: ${{ secrets.GABRIELSIMMER_COM_KEY }}

Cargo.toml

@@ -15,7 +15,7 @@ include_dir = "0.7.3"
 frontmatter = "0.4.0"
 comrak = { version = "0.21.0", default-features = false }
 orgize = { git = "https://git.gmem.ca/arch/orgize.git", branch = "org-images", default-features = false }
-rss = "2.0.6"
+rss = { version = "2.0.6" }
 time = { version = "0.3.31", features = ["parsing", "formatting", "macros", "wasm-bindgen"] }
 
 [profile.release]

src/lib.rs

@@ -27,10 +27,20 @@ struct ProjectConfig {
     experiments: Vec<Project>,
 }
 
-#[derive(Serialize, Deserialize, Debug)]
-struct IndexingItem {
-    name: String,
-    date: u64
+#[derive(Deserialize, Debug)]
+struct R2Object {
+    key: String,
+    size: u64,
+    #[serde(rename = "eTag")]
+    e_tag: String,
+}
+
+#[derive(Debug, Deserialize)]
+struct R2Message {
+    account: String,
+    action: String,
+    bucket: String,
+    object: R2Object,
 }
 
 static DHALL_DIR: Dir<'_> = include_dir!("$CARGO_MANIFEST_DIR/src/dhall");
@@ -197,25 +207,6 @@ async fn main(req: Request, env: Env, _ctx: Context) -> Result<Response> {
             cache.put(key, res.cloned()?).await?;
             Ok(res)
         })
-        .get_async("/api/reindex", |req, ctx| async move {
-            let auth_header = req.headers().get("Authorization")?;
-            if auth_header.is_none() {
-                return Response::empty();
-            }
-            if auth_header.unwrap() != ctx.env.secret("AUTHENTICATION_TOKEN")?.to_string() {
-                return Response::empty();
-            }
-            let bucket = ctx.env.bucket("GABRIELSIMMERCOM_BUCKET")?;
-            let queue = ctx.env.queue("INDEXING")?;
-            let contents = bucket.list().prefix("posts/").execute().await?.objects();
-            for item in &contents {
-                let indexing_item = IndexingItem{
-                    name: item.key(), date: item.uploaded().as_millis() };
-                queue.send(&indexing_item).await?;
-            }
-            let result = format!("queued {} items for reindexing", &contents.len());
-            Response::ok(result)
-        })
         .run(req, env).await
 }
 
@@ -256,20 +247,20 @@ fn header(page_title: &str) -> Markup {
 }
 
 #[event(queue)]
-async fn queue(message_batch: MessageBatch<IndexingItem>, env: Env, _ctx: Context) -> Result<()> {
+async fn queue(message_batch: MessageBatch<R2Message>, env: Env, _ctx: Context) -> Result<()> {
     let index = env.d1("INDEX")?;
     let bucket = env.bucket("GABRIELSIMMERCOM_BUCKET")?;
     let messages = message_batch.messages()?;
     for message in messages {
-        let fname = message.body.name;
+        let fname = message.body.object.key;
         console_log!("got {} to render and index", &fname);
         let file = bucket.get(&fname).execute().await?;
         if file.is_none() {
             console_warn!("did not find a file in bucket for {}", fname)
         }
         let rendered = posts::blog_post(fname.clone(), file.unwrap().body().unwrap().text().await?).unwrap();
-        
-        let statement = index.prepare("INSERT OR IGNORE INTO posts (slug, title, html, date) VALUES (?1, ?2, ?3, ?4)");
+
+        let statement = index.prepare("INSERT INTO posts (slug, title, html, date) VALUES (?1, ?2, ?3, ?4) ON CONFLICT(slug) DO UPDATE SET html=?3;");
         let query = statement.bind(&[
             rendered.slug.into(), rendered.title.into(), rendered.html.into(), rendered.date.into()
         ]).unwrap();