Flow configuration separation

[rmn-boiko]

Description

Closes: #987
#938

Checklist before requesting a review

• Unit and integration tests added
• Compatibility:
  • Network APIs: ✅
  • Workspace layout and metadata: ✅
  • Configuration: ✅
  • Container images: ✅
• Observability:
  • Tracing / metrics: ✅
  • Alerts: ✅
• Documentation:
  • Changelog: ✅
  • Public documentation: ✅
• Downstream effects:
  • kamu-node: ✅
  • kamu-web-ui: ✅
  • release-queue

[zaychenko-sergei]

Prepare a migration to clean old configuration events

[zaychenko-sergei]

This logic looks obsolete. I think we should restore auto-polling flows based on active triggers, the configuration cannot be in disabled state anymore - it's just a set of properties to take with the triggered flow.

  #[tracing::instrument(level = "debug", skip_all)]
    async fn restore_auto_polling_flows_from_configurations(
        &self,
        target_catalog: &Catalog,
        start_time: DateTime<Utc>,
    ) -> Result<(), InternalError> {
        let flow_configuration_service = target_catalog
            .get_one::<dyn FlowConfigurationService>()
            .unwrap();
        let flow_event_store = target_catalog.get_one::<dyn FlowEventStore>().unwrap();

        // Query all enabled flow configurations
        let enabled_configurations: Vec<_> = flow_configuration_service
            .list_enabled_configurations()
            .try_collect()
            .await?;

        // Split configs by those which have a schedule or different rules
        let (schedule_configs, non_schedule_configs): (Vec<_>, Vec<_>) = enabled_configurations
            .into_iter()
            .partition(|config| matches!(config.rule, FlowConfigurationRule::Schedule(_)));

        let scheduling_helper = target_catalog.get_one::<FlowSchedulingHelper>().unwrap();

        // Activate all configs, ensuring schedule configs precedes non-schedule configs
        // (this i.e. forces all root datasets to be updated earlier than the derived)
        //
        // Thought: maybe we need topological sorting by derived relations as well to
        // optimize the initial execution order, but batching rules may work just fine
        for enabled_config in schedule_configs
            .into_iter()
            .chain(non_schedule_configs.into_iter())
        {
            // Do not re-trigger the flow that has already triggered
            let maybe_pending_flow_id = flow_event_store
                .try_get_pending_flow(&enabled_config.flow_key)
                .await?;
            if maybe_pending_flow_id.is_none() {
                scheduling_helper
                    .activate_flow_configuration(
                        start_time,
                        enabled_config.flow_key,
                        enabled_config.rule,
                    )
                    .await?;
            }
        }

        Ok(())
    }