fix: br_me_caged não detecta atualizações do CAGED

pipelines/datasets/br_me_caged/flows.py

@@ -18,9 +18,9 @@
     YearMonth,
 )
 from pipelines.utils.metadata.tasks import (
-    commit_source_update_task,
-    poll_source_for_update_task,
-    register_table_materialization_task,
+    check_source_is_ahead_of_table_task,
+    register_source_coverage_task,
+    sync_table_coverage_task,
 )
 from pipelines.utils.tasks import (
     rename_flow_run_dataset_table,
@@ -46,17 +46,21 @@ def _run_me_caged(
 
     source_last_date = get_source_last_date()
 
-    if not force_run:
-        has_new_data = poll_source_for_update_task(
-            dataset_id=dataset_id,
-            table_id=table_id,
-            source_max_date=source_last_date,
-            env="prod",
-            date_format="%Y-%m",
-        )
-        if not has_new_data:
-            print(f"No updates for table {table_id}!")
-            return
+    register_source_coverage_task(
+        dataset_id=dataset_id,
+        table_id=table_id,
+        source_max_date=source_last_date,
+        env="prod",  # trocar para "dev" ao testar localmente
+        date_format="%Y-%m",
+    )
+
+    if not force_run and not check_source_is_ahead_of_table_task(
+        dataset_id=dataset_id,
+        table_id=table_id,
+        env="prod",  # trocar para "dev" ao testar localmente
+    ):
+        print(f"No updates for table {table_id}!")
+        return
 
     table_last_date = get_table_last_date(dataset_id, table_id)
     _input_dir, output_dir = build_table_paths(table_id)
@@ -104,26 +108,17 @@ def _run_me_caged(
     )
 
     if update_metadata:
-        register_table_materialization_task(
+        sync_table_coverage_task(
             dataset_id=dataset_id,
             table_id=table_id,
             coverage=PartBdpro(
                 date_column=YearMonth(year="ano", month="mes"),
                 date_format=DateFormat.YEAR_MONTH,
             ),
-            env="prod",
+            env="prod",  # trocar para "dev" ao testar localmente
             bq_project="basedosdados",
         )
 
-        if source_last_date is not None:
-            commit_source_update_task(
-                dataset_id=dataset_id,
-                table_id=table_id,
-                source_max_date=source_last_date,
-                env="prod",
-                date_format="%Y-%m",
-            )
-
 
 def _caged_flow(table_id: str, cron: str):
     @flow(