update es-update-job to include db read time

_elasticsearch/index_definitions/media_files_v1.json

@@ -39,6 +39,8 @@
       "user_deleted_at" : { "type" : "date" },
       "mod_deleted_at" : { "type" : "date" },
 
+      "database_read_time" : { "type" : "date" },
+
       "is_featured" : {
         "type": "boolean",
         "index": true

_elasticsearch/index_definitions/model_weights_v1.json

@@ -37,6 +37,8 @@
       "user_deleted_at" : { "type" : "date" },
       "mod_deleted_at" : { "type" : "date" },
 
+      "database_read_time" : { "type" : "date" },
+
       "is_deleted" : {
         "type": "boolean",
         "index": true

crates/lib/elasticsearch_schema/src/documents/media_file_document.rs

@@ -64,6 +64,9 @@ pub struct MediaFileDocument {
   /// Whether the file is featured.
   pub is_featured: bool,
 
+  /// When we read the record from the database
+  pub database_read_time: DateTime<Utc>,
+
   /// Calculated as "either user or mod deleted"
   pub is_deleted: bool,
 }

crates/lib/elasticsearch_schema/src/documents/model_weight_document.rs

@@ -63,6 +63,9 @@ pub struct ModelWeightDocument {
   pub user_deleted_at: Option<DateTime<Utc>>,
   pub mod_deleted_at: Option<DateTime<Utc>>,
 
+  /// When we read the record from the database
+  pub database_read_time: DateTime<Utc>,
+
   /// Calculated as "either user or mod deleted"
   pub is_deleted: bool,
 }

crates/lib/mysql_queries/src/queries/media_files/list/list_media_files_for_elastic_search_backfill_using_cursor.rs

@@ -112,6 +112,8 @@ pub struct MediaFileForElasticsearchRecord {
 
   pub user_deleted_at: Option<DateTime<Utc>>,
   pub mod_deleted_at: Option<DateTime<Utc>>,
+
+  pub database_read_time: DateTime<Utc>,
 }
 
 pub struct ListArgs<'a> {
@@ -197,6 +199,7 @@ pub async fn list_media_files_for_elastic_search_backfill_using_cursor(
           updated_at: record.updated_at,
           user_deleted_at: record.user_deleted_at,
           mod_deleted_at: record.mod_deleted_at,
+          database_read_time: record.database_read_time,
         }
       })
       .collect::<Vec<MediaFileForElasticsearchRecord>>())
@@ -277,7 +280,9 @@ SELECT
     m.created_at,
     m.updated_at,
     m.user_deleted_at,
-    m.mod_deleted_at
+    m.mod_deleted_at,
+
+    NOW() as database_read_time
 
 FROM media_files as m
 
@@ -434,6 +439,8 @@ struct RawRecord {
 
   pub user_deleted_at: Option<DateTime<Utc>>,
   pub mod_deleted_at: Option<DateTime<Utc>>,
+
+  pub database_read_time: DateTime<Utc>,
 }
 
 impl FromRow<'_, MySqlRow> for RawRecord {
@@ -509,6 +516,8 @@ impl FromRow<'_, MySqlRow> for RawRecord {
       updated_at: row.try_get("updated_at")?,
       user_deleted_at: row.try_get("user_deleted_at")?,
       mod_deleted_at: row.try_get("mod_deleted_at")?,
+
+      database_read_time: row.try_get("database_read_time")?,
     })
   }
 }

crates/lib/mysql_queries/src/queries/model_weights/batch_get/batch_get_model_weights_for_elastic_search_backfill.rs

@@ -56,6 +56,8 @@ pub struct ModelWeightForElasticsearchRecord {
 
   pub user_deleted_at: Option<DateTime<Utc>>,
   pub mod_deleted_at: Option<DateTime<Utc>>,
+
+  pub database_read_time: DateTime<Utc>,
 }
 
 pub async fn batch_get_model_weights_for_elastic_search_backfill<'e, 'c, E>(
@@ -111,6 +113,7 @@ pub async fn batch_get_model_weights_for_elastic_search_backfill<'e, 'c, E>(
           updated_at: model.updated_at,
           user_deleted_at: model.user_deleted_at,
           mod_deleted_at: model.mod_deleted_at,
+          database_read_time: model.database_read_time,
         }
       })
       .collect::<Vec<ModelWeightForElasticsearchRecord>>())
@@ -162,7 +165,9 @@ SELECT
     w.created_at,
     w.updated_at,
     w.user_deleted_at,
-    w.mod_deleted_at
+    w.mod_deleted_at,
+
+    NOW() as database_read_time
 
 FROM model_weights as w
 
@@ -242,6 +247,8 @@ struct RawRecord {
 
   pub user_deleted_at: Option<DateTime<Utc>>,
   pub mod_deleted_at: Option<DateTime<Utc>>,
+
+  pub database_read_time: DateTime<Utc>,
 }
 
 // NB(bt,2023-12-05): There's an issue with type hinting in the `as` clauses with QueryBuilder (or
@@ -296,6 +303,7 @@ impl FromRow<'_, MySqlRow> for RawRecord {
       updated_at: row.try_get("updated_at")?,
       user_deleted_at: row.try_get("user_deleted_at")?,
       mod_deleted_at: row.try_get("mod_deleted_at")?,
+      database_read_time: row.try_get("database_read_time")?,
     })
   }
 }

crates/service/job/es-update-job/src/tasks/update_all_model_weights/update_all_model_weights.rs

@@ -151,6 +151,8 @@ async fn create_document_from_record(elasticsearch: &Elasticsearch, record: Mode
     user_deleted_at: record.user_deleted_at,
     mod_deleted_at: record.mod_deleted_at,
 
+    database_read_time: record.database_read_time,
+
     is_deleted,
   };
 

crates/service/job/es-update-job/src/tasks/update_engine_media_files/update_engine_media_files.rs

@@ -156,6 +156,8 @@ async fn create_document_from_record(elasticsearch: &Elasticsearch, record: Medi
     user_deleted_at: record.user_deleted_at,
     mod_deleted_at: record.mod_deleted_at,
 
+    database_read_time: record.database_read_time,
+
     is_deleted,
   };