fix(consumer): a few backfill fixes

- move file creation
- log more error detail in backfill_jobs
- move the annoying unknown record error to debug

consumer/src/backfill/downloader.rs

···
             Ok(Some(did_doc)) => {
                 let Some(service) = did_doc.find_service_by_id(PDS_SERVICE_ID) else {
                     tracing::warn!("bad DID doc for {did}");
-                    db::backfill_job_write(&mut conn, &did, "failed.resolve")
+                    db::backfill_job_write(&mut conn, &did, "failed.resolve.did_svc")
                         .await
                         .unwrap();
                     continue;
···
                 }
             }
             Ok(None) => {
-                tracing::warn!(did, "bad DID doc");
+                tracing::warn!(did, "bad/missing DID doc");
                 db::actor_set_sync_status(&mut conn, &did, ActorSyncState::Dirty, Utc::now())
                     .await
                     .unwrap();
-                db::backfill_job_write(&mut conn, &did, "failed.resolve")
+                db::backfill_job_write(&mut conn, &did, "failed.resolve.did_doc")
                     .await
                     .unwrap();
             }
···
                 db::actor_set_sync_status(&mut conn, &did, ActorSyncState::Dirty, Utc::now())
                     .await
                     .unwrap();
-                db::backfill_job_write(&mut conn, &did, "failed.resolve")
+                db::backfill_job_write(&mut conn, &did, "failed.resolve.did")
                     .await
                     .unwrap();
             }
···
                 Ok(false) => continue,
                 Err(e) => {
                     tracing::error!(pds, did, "failed to check repo status: {e}");
-                    db::backfill_job_write(&mut conn, &did, "failed.resolve")
+                    db::backfill_job_write(&mut conn, &did, "failed.resolve.status")
                         .await
                         .unwrap();
                     continue;
···
             if let Some(handle) = maybe_handle {
                 if let Err(e) = resolve_and_set_handle(&conn, &resolver, &did, &handle).await {
                     tracing::error!(pds, did, "failed to resolve handle: {e}");
-                    db::backfill_job_write(&mut conn, &did, "failed.resolve")
+                    db::backfill_job_write(&mut conn, &did, "failed.resolve.handle")
                         .await
                         .unwrap();
                 }
···
     pds: &str,
     did: &str,
 ) -> eyre::Result<Option<(i32, i32)>> {
-    let mut file = tokio::fs::File::create_new(tmp_dir.join(did)).await?;
-
     let res = http
         .get(format!("{pds}/xrpc/com.atproto.sync.getRepo?did={did}"))
         .send()
         .await?
         .error_for_status()?;
+
+    let mut file = tokio::fs::File::create_new(tmp_dir.join(did)).await?;
 
     let headers = res.headers();
     let ratelimit_rem = header_to_int(headers, "ratelimit-remaining");

consumer/src/backfill/repo.rs

···
 use super::{
-    types::{CarCommitEntry, CarEntry},
+    types::{CarCommitEntry, CarEntry, CarRecordEntry},
     CopyStore,
 };
 use crate::indexer::records;
···
             CarEntry::Commit(_) => {
                 tracing::warn!("got commit entry that was not in root")
             }
-            CarEntry::Record(record) => {
+            CarEntry::Record(CarRecordEntry::Known(record)) => {
                 if let Some(path) = mst_nodes.remove(&cid) {
                     record_index(t, rc, &mut copies, &mut deltas, repo, &path, cid, record).await?;
                 } else {
                     records.insert(cid, record);
                 }
+            }
+            CarEntry::Record(CarRecordEntry::Other { ty }) => {
+                tracing::debug!("repo contains unknown record type: {ty} ({cid})");
             }
             CarEntry::Mst(mst) => {
                 let mut out = Vec::with_capacity(mst.e.len());

consumer/src/backfill/types.rs

···
 pub enum CarEntry {
     Mst(CarMstEntry),
     Commit(CarCommitEntry),
-    Record(RecordTypes),
+    Record(CarRecordEntry),
 }
 
 #[derive(Debug, Deserialize)]
···
     pub rev: String,
     pub prev: Option<Cid>,
     pub sig: ByteBuf,
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(untagged)]
+pub enum CarRecordEntry {
+    Known(RecordTypes),
+    Other {
+        #[serde(rename = "$type")]
+        ty: String,
+    },
 }
 
 #[derive(Debug, Deserialize)]

consumer/src/db/backfill.rs

···
     status: &str,
 ) -> PgExecResult {
     conn.execute(
-        "INSERT INTO backfill_jobs (did, status) VALUES ($1, $2)",
+        "INSERT INTO backfill_jobs (did, status) VALUES ($1, $2) ON CONFLICT (did) DO UPDATE SET status = $2, updated_at = NOW()",
         &[&did, &status],
     )
     .await