"""
Amazon Music Ingestion Workflow.

Ingests Amazon Music (Unlimited, Prime, AdSupported) data
and loads into fact analytics.
"""
import datetime
import os

from garcon.param import StaticParam
from garcon_contrib.aws import garcon_sns
from garcon_contrib.dynamo_feed_status import garcon_feed_status

from feed_ingestion.flows import base
from feed_ingestion.flows.amazon_music import config
from feed_ingestion.flows.amazon_music import generators
from feed_ingestion.flows.amazon_music import tasks
from feed_ingestion.tasks import feed_status_tasks
from feed_ingestion.tasks import load_fact_tables_tasks_sf
from feed_ingestion.tasks import load_raw_table_tasks_sf
from feed_ingestion.tasks import overall_status_tasks
from feed_ingestion.tasks import s3_tasks
from feed_ingestion.tasks import validate_raw_data_tasks_sf
from feed_ingestion.util.aws import s3
from feed_ingestion.util.jenkins.tasks import build_jenkins_dbt


class Flow(base.FlowBase, base.FlowConfigMixin):
    """Amazon Music Flow class.

    This flow downloads the Amazon Music data and loads it into
    staging_raw and fact_analytics/fact_analytics_error tables.
    """

    def __init__(self):
        """Initialize a Amazon Music flow."""
        super(Flow, self).__init__(config.feed_name, config.feed_version)
        self.timeout = 60 * 60 * 4  # override default to 4 hours
        self.sns_topic = os.environ.get(
            'FEED_INGESTION_SNS_TOPIC',
            'arn:aws:sns:us-east-1:437795906767:dev-swf-feed-ingestion:'
            'e17576f8-b905-45a9-b863-0eaceca54f4d')

    def decider(self, schedule):
        """Orchestrate an Amazon Music workflow.

        Args:
            schedule (callable): The scheduler method.
        """
        bootstrap = schedule(
            'bootstrap', self.bootstrap)

        if (bootstrap.result.get('bootstrap_feed.source')
                == config.SOURCE_DATAPULSE):
            if bootstrap.result.get('bootstrap_feed.stop') is True:
                return
            self._decider_datapulse(schedule, bootstrap)
            return

        next_step_requires = [bootstrap]

        if not bootstrap.result.get('bootstrap_feed.use_s3'):
            remove_stale_stage_files = schedule(
                'remove_stale_stage_files',
                self.remove_stale_stage_files,
                requires=[bootstrap])
            next_step_requires = [remove_stale_stage_files]

        if bootstrap.result.get('bootstrap_feed.stop') is True:
            return

        if bootstrap.result.get('bootstrap_feed.licensor') == 'sme':
            map_expected_files_to_sme = schedule(
                'map_expected_files_to_sme',
                self.map_expected_files_to_sme,
                requires=next_step_requires)

            if not map_expected_files_to_sme.result.get(
                    'map_expected_files_to_sme.has_files'):
                return

            sme_grab_and_clean_files = schedule(
                'sme_grab_and_clean_files',
                self.sme_grab_and_clean_files,
                requires=[map_expected_files_to_sme])
            next_step_requires = [sme_grab_and_clean_files]
        elif bootstrap.result.get('bootstrap_feed.licensor') == 'awal':
            if bootstrap.result.get('bootstrap_feed.use_s3'):
                awal_grab_drop_files_s3 = schedule(
                    'awal_grab_drop_files_s3',
                    self.awal_grab_drop_files_s3,
                    requires=next_step_requires)
                next_step_requires = [awal_grab_drop_files_s3]
            else:
                awal_grab_drop_files = schedule(
                    'awal_grab_drop_files',
                    self.awal_grab_drop_files,
                    requires=next_step_requires)
                next_step_requires = [awal_grab_drop_files]
        elif bootstrap.result.get('bootstrap_feed.licensor') == 'altafonte':
            if bootstrap.result.get('bootstrap_feed.use_s3'):
                altafonte_grab_drop_files_s3 = schedule(
                    'altafonte_grab_drop_files_s3',
                    self.altafonte_grab_drop_files_s3,
                    requires=next_step_requires)
                next_step_requires = [altafonte_grab_drop_files_s3]
            else:
                altafonte_grab_drop_files = schedule(
                    'altafonte_grab_drop_files',
                    self.altafonte_grab_drop_files,
                    requires=next_step_requires)
                next_step_requires = [altafonte_grab_drop_files]
        else:
            grab_drop_files = schedule(
                'grab_drop_files',
                self.grab_drop_files,
                requires=next_step_requires)
            next_step_requires = [grab_drop_files]

        update_feed_s3_file_status = schedule(
            'update_feed_s3_file_status',
            self.update_feed_s3_file_status,
            requires=next_step_requires)

        if (self._is_update_feed_status_available(update_feed_s3_file_status)
                and not bootstrap.result.get('bootstrap_feed.soft_reload')):
            return

        set_feed_status_downloaded = schedule(
            'set_feed_status_downloaded',
            self.set_status_to_downloaded,
            requires=[update_feed_s3_file_status])

        if bootstrap.result.get('bootstrap_feed.licensor') == 'sme':
            # clean for sme was performed in sme_grab_and_clean_files task
            next_step_requires = [update_feed_s3_file_status]
        elif bootstrap.result.get('bootstrap_feed.licensor') == 'awal':
            next_step_requires = [set_feed_status_downloaded]
            if not bootstrap.result.get('bootstrap_feed.use_s3'):
                convert_zip_to_gzip_on_s3 = schedule(
                    'convert_zip_to_gzip_on_s3',
                    self.convert_zip_to_gzip_on_s3,
                    requires=[set_feed_status_downloaded])
                next_step_requires = [convert_zip_to_gzip_on_s3]
        elif bootstrap.result.get('bootstrap_feed.licensor') == 'altafonte':
            convert_zip_to_gzip_on_s3 = schedule(
                'convert_zip_to_gzip_on_s3',
                self.convert_zip_to_gzip_on_s3,
                requires=[set_feed_status_downloaded])
            next_step_requires = [convert_zip_to_gzip_on_s3]
        else:
            if bootstrap.result.get('bootstrap_feed.soft_reload'):
                next_step_requires = [set_feed_status_downloaded]
            else:
                clean_files = schedule(
                    'clean_files',
                    self.clean_files,
                    requires=[set_feed_status_downloaded])
                next_step_requires = [clean_files]

        create_temp_staging_raw_tables = schedule(
            'create_temp_staging_raw_tables',
            self.create_temp_staging_raw_tables,
            requires=next_step_requires)

        load_temp_staging_raw_tables = schedule(
            'load_temp_staging_raw_tables',
            self.load_temp_staging_raw_tables,
            requires=[create_temp_staging_raw_tables])

        clean_staging_raw_table = schedule(
            'clean_staging_raw_table',
            self.clean_staging_raw_table,
            requires=[load_temp_staging_raw_tables])

        load_staging_raw_table = schedule(
            'load_staging_raw_table',
            self.load_staging_raw_table,
            requires=[clean_staging_raw_table])

        mark_staging_raw_table_tasks_complete = schedule(
            'mark_staging_raw_table_tasks_complete',
            self.mark_staging_raw_table_tasks_complete,
            requires=[load_staging_raw_table])

        set_status_populated_raw_table = schedule(
            'set_status_populated_raw_table',
            self.set_status_populated_raw_table,
            requires=[mark_staging_raw_table_tasks_complete])

        drop_temp_staging_raw_tables = schedule(
            'drop_temp_staging_raw_tables',
            self.drop_temp_staging_raw_table,
            requires=[set_status_populated_raw_table])

        if bootstrap.result.get('bootstrap_feed.populate_only'):
            return

        load_aggregated_table = schedule(
            'load_aggregated_table',
            self.load_aggregated_table,
            requires=[set_status_populated_raw_table])

        update_dim_tables = schedule(
            'update_dim_tables',
            self.update_dim_tables,
            requires=[drop_temp_staging_raw_tables])

        if update_dim_tables.result.get(
                'update_dim_tables.sns_report_subject', False):
            schedule(
                'send_dimension_tables_update_report_sns_notification',
                self.send_dimension_tables_update_report_sns_notification,
                requires=[update_dim_tables])

        load_staging_fact_table = schedule(
            'load_staging_fact_table',
            self.load_staging_fact_table,
            requires=[update_dim_tables, load_aggregated_table])

        load_aggregated_skips_and_saves = schedule(
            'load_aggregated_skips_and_saves',
            self.load_aggregated_skips_and_saves,
            requires=[load_staging_fact_table])

        load_fact_tables = schedule(
            'load_fact_tables',
            self.load_fact_tables,
            requires=[load_aggregated_skips_and_saves,
                      load_staging_fact_table])

        clean_up = schedule(
            'clean_up',
            self.clean_up,
            requires=[load_fact_tables])

        build_jenkins_dbt = schedule(
            'build_jenkins_dbt',
            self.build_jenkins_dbt,
            requires=[clean_up])

        if self.sns_topic:
            monitor_drop_location = schedule(
                'monitor_drop_location',
                self.monitor_drop_location,
                requires=[build_jenkins_dbt])

            if monitor_drop_location.result.get(
                    'monitor_drop_location.unexpected_files_message') \
                    is not None:
                schedule(
                    'send_unexpected_files_sns_notification',
                    self.send_unexpected_files_sns_notification,
                    requires=[monitor_drop_location])

            if monitor_drop_location.result.get(
                    'monitor_drop_location.stability_files_message') \
                    is not None:
                schedule(
                    'send_stability_report_sns_notification',
                    self.send_stability_report_sns_notification,
                    requires=[monitor_drop_location])

    def _decider_datapulse(self, schedule, bootstrap):
        """DAG when source=datapulse: skip downloads, load from Snowflake.

        TODO: When we decomission SFTP source and rely solely on datapulse,
        we can remove the other branches in this method and simplify the
        decider.

        Args:
            schedule (callable): The scheduler method.
            bootstrap: The bootstrap activity result.
        """
        clean_staging_raw_table = schedule(
            'clean_staging_raw_table',
            self.clean_staging_raw_table,
            requires=[bootstrap])

        load_staging_raw_datapulse = schedule(
            'load_staging_raw_datapulse',
            self.load_staging_raw_datapulse,
            requires=[clean_staging_raw_table])

        mark_staging_raw_table_tasks_complete = schedule(
            'mark_staging_raw_table_tasks_complete',
            self.mark_staging_raw_table_tasks_complete,
            requires=[load_staging_raw_datapulse])

        set_status_populated_raw_table = schedule(
            'set_status_populated_raw_table',
            self.set_status_populated_raw_table,
            requires=[mark_staging_raw_table_tasks_complete])

        if bootstrap.result.get('bootstrap_feed.populate_only'):
            return

        load_aggregated_table = schedule(
            'load_aggregated_table',
            self.load_aggregated_table,
            requires=[set_status_populated_raw_table])

        update_dim_tables = schedule(
            'update_dim_tables',
            self.update_dim_tables,
            requires=[set_status_populated_raw_table])

        if update_dim_tables.result.get(
                'update_dim_tables.sns_report_subject', False):
            schedule(
                'send_dimension_tables_update_report_sns_notification',
                self.send_dimension_tables_update_report_sns_notification,
                requires=[update_dim_tables])

        load_staging_fact_table = schedule(
            'load_staging_fact_table',
            self.load_staging_fact_table,
            requires=[update_dim_tables, load_aggregated_table])

        load_aggregated_skips_and_saves = schedule(
            'load_aggregated_skips_and_saves',
            self.load_aggregated_skips_and_saves,
            requires=[load_staging_fact_table])

        load_fact_tables = schedule(
            'load_fact_tables',
            self.load_fact_tables,
            requires=[load_aggregated_skips_and_saves,
                      load_staging_fact_table])

        clean_up = schedule(
            'clean_up',
            self.clean_up,
            requires=[load_fact_tables])

        schedule(
            'build_jenkins_dbt',
            self.build_jenkins_dbt,
            requires=[clean_up])

    def workflow_id(self, initial_context):
        """Generate workflow id.

        Args:
            initial_context (dict): The initial context for the flow.
        Returns:
            str: A unique identifier for a workflow being executed.
        """
        report_name = initial_context['report_name']
        licensor = initial_context['licensor']
        flow_name = '_'.join([self.name, report_name])

        if 'context_date' not in initial_context:
            date = datetime.datetime.today().strftime(config.DATE_FORMAT)
        else:
            date = initial_context['context_date']

        return '{flow_name}-{licensor}-{date}'.format(
            flow_name=flow_name,
            licensor=licensor,
            date=date)

    def contextified_feed_name(self, context):
        """Get feed_name in context.

        Args:
            context (dict): The context of the flow.
        Returns:
            str: Contextified feed name.
        """
        assert 'report_name' in context, 'There is no report_name in context'
        assert context['report_name'] in config.reports, \
            'There is no such report_name in config file'
        assert 'licensor' in context, 'There is no licensor in context'
        assert context['licensor'] in config.licensors, \
            'There is no such licensor in config file'
        return '_'.join([
            self.feed_name, context['licensor'], context['report_name'],
        ])

    @property
    def bootstrap(self):
        """Return a new garson Activity which bootstraps the flow.

        Returns:
            garcon.Activity: A bootstrap activity
        """
        return self.create(
            name='bootstrap',
            tasks=base.SyncRunner(
                tasks.bootstrap.fill(
                    namespace='bootstrap_feed',
                    date='context_date',
                    report_name='report_name',
                    reload='reload',
                    soft_reload='soft_reload',
                    countries='countries',
                    snowflake_error_limit='snowflake_error_limit',
                    licensor='licensor',
                    use_s3='use_s3',
                    populate_only='populate_only',
                    source='source',
                    # This 79 char limit really makes code look better:
                    snowflake_error_on_column_count_mismatch=(
                        'snowflake_error_on_column_count_mismatch'),
                    extract_original_filename=(
                        'extract_original_filename')
                )))

    @property
    def remove_stale_stage_files(self):
        """Remove stale files in stage directory."""
        return self.create(
            name='remove_stale_stage_files',
            tasks=base.SyncRunner(
                tasks.remove_stale_stage_files.fill(
                    report_name='bootstrap_feed.report_name',
                    licensor='bootstrap_feed.licensor',
                )))

    @property
    def grab_drop_files(self):
        """Grab files and upload to archive bucket on S3."""
        return self.create(
            name='grab_drop_files',
            generators=[generators.drop_files_generator],
            tasks=base.AsyncRunner(
                s3_tasks.copy_file.fill(
                    namespace='copy_file',
                    source_bucket_name=StaticParam(config.drop_bucket),
                    source_key_name='source_key_name',
                    destination_bucket_name=StaticParam(config.archive_bucket),
                    destination_key_name='destination_key_name',
                    replace='bootstrap_feed.replace_archive_files'),
                max_workers=4))

    @property
    def awal_grab_drop_files(self):
        """Grab files and upload to archive bucket on S3."""
        return self.create(
            schedule_to_start=10000,
            name='awal_grab_drop_files',
            generators=[generators.drop_files_generator],
            tasks=base.SyncRunner(
                tasks.awal_grab_drop_files.fill(
                    namespace='awal_grab_drop_files',
                    date='bootstrap_feed.date',
                    report_name='report_name',
                    s3_archive_path='bootstrap_feed.archive_bucket',
                    destination_key_name='destination_key_name')))

    @property
    def awal_grab_drop_files_s3(self):
        """Grab files and upload to archive bucket on S3."""
        return self.create(
            schedule_to_start=10000,
            name='awal_grab_drop_files_s3',
            generators=[generators.awal_s3_files_generator],
            tasks=base.SyncRunner(
                tasks.awal_grab_drop_files_s3.fill(
                    namespace='awal_grab_drop_files_s3',
                    date='bootstrap_feed.date',
                    source_key_name='source_key_name',
                    destination_key_name='destination_key_name'
                )))

    @property
    def altafonte_grab_drop_files(self):
        """Grab files and upload to archive bucket on S3."""
        return self.create(
            schedule_to_start=10000,
            name='altafonte_grab_drop_files',
            generators=[generators.drop_files_generator],
            tasks=base.SyncRunner(
                tasks.altafonte_grab_drop_files.fill(
                    namespace='altafonte_grab_drop_files',
                    date='bootstrap_feed.date',
                    report_name='report_name',
                    s3_archive_path='bootstrap_feed.archive_bucket',
                    destination_key_name='destination_key_name')))

    @property
    def altafonte_grab_drop_files_s3(self):
        """Grab files and upload to archive bucket on S3."""
        return self.create(
            schedule_to_start=10000,
            name='altafonte_grab_drop_files_s3',
            generators=[generators.altafonte_s3_files_generator],
            tasks=base.SyncRunner(
                tasks.altafonte_grab_drop_files_s3.fill(
                    namespace='altafonte_grab_drop_files_s3',
                    date='bootstrap_feed.date',
                    source_key_name='source_key_name',
                    destination_key_name='destination_key_name'
                )))

    @property
    def map_expected_files_to_sme(self):
        """Map files in SME S3 bucket to expected_files."""
        return self.create(
            name='map_expected_files_to_sme',
            tasks=base.AsyncRunner(
                tasks.map_expected_files_to_sme.fill(
                    namespace='map_expected_files_to_sme',
                    s3_drop_location='bootstrap_feed.s3_drop_location',
                    date='bootstrap_feed.date',
                    expected_files='bootstrap_feed.expected_files',
                )
            )
        )

    @property
    def sme_grab_and_clean_files(self):
        """Grab files from SME, clean and upload to archive bucket on S3."""
        return self.create(
            name='sme_grab_and_clean_files',
            generators=[generators.sme_drop_files_generator],
            tasks=base.AsyncRunner(
                tasks.sme_grab_and_clean.fill(
                    namespace='sme_grab_and_clean',
                    source_bucket_name=StaticParam(config.sme_drop_bucket),
                    source_key_name='source_key_name',
                    destination_bucket_name=StaticParam(config.archive_bucket),
                    destination_key_name='destination_key_name',
                    date='bootstrap_feed.date',
                    feed_name='bootstrap_feed.feed_name',
                    report_name='report_name',
                ),
                max_workers=4))

    @property
    def update_feed_s3_file_status(self):
        """Update status of downloaded files."""
        return self.create(
            name='update_feed_s3_file_status',
            tasks=base.SyncRunner(
                feed_status_tasks.update_feed_s3_file_status.fill(
                    namespace='update_feed_s3_file_status',
                    s3_path='bootstrap_feed.archive_bucket',
                    file_names='bootstrap_feed.expected_files',
                    feed_name='bootstrap_feed.feed_name',
                    date='bootstrap_feed.date')))

    @property
    def set_status_to_downloaded(self):
        """Set overall feed status to DOWNLOADED."""
        return self.create(
            name='set_status_to_downloaded',
            tasks=base.SyncRunner(
                overall_status_tasks.set_overall_status.fill(
                    namespace='set_status_to_downloaded',
                    feed_name='bootstrap_feed.feed_name',
                    date='bootstrap_feed.date',
                    status=StaticParam(
                        garcon_feed_status.STATUS_DOWNLOADED))))

    @property
    def convert_zip_to_gzip_on_s3(self):
        """Convert zip archive to gzip on s3."""
        return self.create(
            name='convert_zip_to_gzip_on_s3',
            generators=[generators.zip_to_gzip_generator],
            tasks=base.SyncRunner(
                s3.convert_zip_to_gzip_on_s3.fill(
                    namespace='convert_zip_to_gzip_on_s3',
                    date='bootstrap_feed.date',
                    zip_s3_path='zip_s3_path',
                    gz_s3_path='gz_s3_path',
                    extract_original_filename=(
                        'bootstrap_feed.extract_original_filename'))))

    @property
    def clean_files(self):
        """Clean and upload files in format accepted by Snowflake."""
        return self.create(
            name='clean_files',
            generators=[generators.drop_files_generator],
            tasks=base.SyncRunner(
                tasks.download_and_clean.fill(
                    namespace='clean_files',
                    date='bootstrap_feed.date',
                    file_key='destination_key_name',
                    archive_bucket=StaticParam(config.archive_bucket),
                    archive_bucket_path='bootstrap_feed.archive_bucket',
                    report_name='report_name')))

    @property
    def create_temp_staging_raw_tables(self):
        """Create temp staging raw table."""
        return self.create(
            name='create_temp_staging_raw_tables',
            generators=[generators.temp_staging_tables_generator],
            tasks=base.SyncRunner(
                load_raw_table_tasks_sf.create_temp_staging_raw_table.fill(
                    namespace='create_temp_staging_raw_tables',
                    date='bootstrap_feed.date',
                    sfdb_params='sfdb_params',
                    feed_name='bootstrap_feed.feed_name',
                    secrets_path=StaticParam(config.secrets_path),
                    temp_staging_raw_table='temp_staging_raw_table',
                    kwargs='kwargs')))

    @property
    def load_temp_staging_raw_tables(self):
        """Load temp staging raw tables."""
        return self.create(
            name='load_temp_staging_raw_tables',
            generators=[generators.temp_staging_tables_generator],
            schedule_to_start=48000,
            tasks=base.SyncRunner(
                validate_raw_data_tasks_sf.load_temp_staging_raw_table.fill(
                    namespace='load_temp_staging_raw_tables',
                    aws=StaticParam(self.conf_aws),
                    date='bootstrap_feed.date',
                    sfdb_params='sfdb_params',
                    feed_name='bootstrap_feed.feed_name',
                    secrets_path=StaticParam(config.secrets_path),
                    kwargs='kwargs',
                    key_dir='key_dir',
                    temp_staging_raw_table='temp_staging_raw_table',
                    error_limit='bootstrap_feed.snowflake_error_limit',
                    # This 79 char limit really makes code look better:
                    snowflake_error_on_column_count_mismatch=(
                        'bootstrap_feed.'
                        'snowflake_error_on_column_count_mismatch')
                )))

    @property
    def clean_staging_raw_table(self):
        """Clean staging raw table for date."""
        return self.create(
            name='clean_staging_raw_table',
            tasks=base.SyncRunner(
                load_raw_table_tasks_sf.clean_staging_raw_table.fill(
                    namespace='clean_staging_raw_table',
                    date='bootstrap_feed.date',
                    feed_name='bootstrap_feed.feed_name',
                    secrets_path=StaticParam(config.secrets_path),
                    sfdb_params='sfdb_params',
                    staging_raw_table='bootstrap_feed.staging_raw_table',
                    kwargs='bootstrap_feed.common_kwargs')))

    @property
    def load_staging_raw_table(self):
        """Load staging raw table."""
        return self.create(
            name='load_staging_raw_table',
            generators=[generators.load_staging_raw_table_generator],
            tasks=base.SyncRunner(
                load_raw_table_tasks_sf.load_staging_raw_table.fill(
                    namespace='load_staging_raw_table',
                    date='bootstrap_feed.date',
                    staging_raw_table='bootstrap_feed.staging_raw_table',
                    sfdb_params='sfdb_params',
                    feed_name='bootstrap_feed.feed_name',
                    secrets_path=StaticParam(config.secrets_path),
                    clean=StaticParam('False'),
                    set_complete=StaticParam('False'),
                    kwargs='kwargs')))

    @property
    def load_staging_raw_datapulse(self):
        """Load staging_raw table from amazon_datapulse staging tables."""
        return self.create(
            name='load_staging_raw_datapulse',
            schedule_to_start=48000,
            tasks=base.SyncRunner(
                tasks.load_staging_raw_from_datapulse.fill(
                    namespace='load_staging_raw_datapulse',
                    date='bootstrap_feed.date',
                    report_name='report_name',
                    licensor='bootstrap_feed.licensor',
                    feed_name='bootstrap_feed.feed_name',
                    secrets_path=StaticParam(config.secrets_path),
                    sfdb_params='sfdb_params')))

    @property
    def mark_staging_raw_table_tasks_complete(self):
        """Set the complete staging_raw_table_tasks to feed."""
        return self.create(
            name='mark_staging_raw_table_tasks_complete',
            tasks=base.SyncRunner(
                (load_raw_table_tasks_sf.mark_staging_raw_table_tasks_complete.
                 fill(
                    namespace='mark_staging_raw_table_tasks_complete',
                    date='bootstrap_feed.date',
                    feed_name='bootstrap_feed.feed_name'))))

    @property
    def set_status_populated_raw_table(self):
        """Set the overall feed status to populated_raw_table."""
        return self.create(
            name='set_status_populated_raw_table',
            tasks=base.SyncRunner(
                overall_status_tasks.set_overall_status.fill(
                    namespace='set_status_populated_raw_table',
                    feed_name='bootstrap_feed.feed_name',
                    date='bootstrap_feed.date',
                    status=StaticParam(
                        garcon_feed_status.STATUS_POPULATED_RAW_TABLE))))

    @property
    def drop_temp_staging_raw_table(self):
        """Drop the temporary staging tables."""
        return self.create(
            name='drop_temp_staging_raw_table',
            generators=[generators.temp_staging_tables_generator],
            schedule_to_start=48000,
            tasks=base.SyncRunner(
                load_raw_table_tasks_sf.drop_temp_staging_raw_table.fill(
                    namespace='drop_temp_staging_table',
                    temp_staging_raw_table='temp_staging_raw_table',
                    feed_name='bootstrap_feed.feed_name',
                    secrets_path=StaticParam(config.secrets_path),
                    sfdb_params='sfdb_params')))

    @property
    def update_dim_tables(self):
        """Update dimension tables."""
        return self.create(
            name='update_dim_tables',
            schedule_to_start=48000,
            tasks=base.SyncRunner(
                load_fact_tables_tasks_sf.update_dim_tables.fill(
                    namespace='update_dim_tables',
                    feed_name='bootstrap_feed.feed_name',
                    secrets_path=StaticParam(config.secrets_path),
                    date='bootstrap_feed.date',
                    sfdb_params='sfdb_params',
                    kwargs='bootstrap_feed.dimension_tables')))

    @property
    def send_dimension_tables_update_report_sns_notification(self):
        """Send SNS message with dimension tables update report."""
        return self.create(
            name='send_dimension_tables_update_report_sns_notification',
            tasks=base.SyncRunner(
                garcon_sns.sns_publish_message.fill(
                    topic=StaticParam(
                        config.dimension_tables['sns_topic']),
                    message='update_dim_tables.sns_report_message',
                    subject='update_dim_tables.sns_report_subject')))

    @property
    def load_aggregated_table(self):
        """Activity to load aggregated staging raw table."""
        return self.create(
            name='load_aggregated_table',
            tasks=base.SyncRunner(
                tasks.load_aggregated_table.fill(
                    namespace='load_aggregated_table',
                    report_name='report_name',
                    feed_name='bootstrap_feed.feed_name',
                    secrets_path=StaticParam(config.secrets_path),
                    date='bootstrap_feed.date',
                    sfdb_params='sfdb_params',
                    licensor='bootstrap_feed.licensor',
                    source='bootstrap_feed.source')))

    @property
    def load_staging_fact_table(self):
        """Activity to load staging_fact_analytics_{feed_name}_{datestamp}."""
        return self.create(
            name='unload_fact_data',
            tasks=base.SyncRunner(
                load_fact_tables_tasks_sf.create_staging_fact.fill(
                    namespace='create_staging_fact',
                    feed_name='bootstrap_feed.feed_name',
                    secrets_path=StaticParam(config.secrets_path),
                    date='bootstrap_feed.date',
                    sfdb_params='sfdb_params'),
                load_fact_tables_tasks_sf.load_staging_fact.fill(
                    namespace='load_staging_fact',
                    feed_name='bootstrap_feed.feed_name',
                    secrets_path=StaticParam(config.secrets_path),
                    date='bootstrap_feed.date',
                    sfdb_params='sfdb_params',
                    kwargs='bootstrap_feed.common_kwargs')))

    @property
    def load_aggregated_skips_and_saves(self):
        """Activity to load load_aggregated_skips_and_saves."""
        return self.create(
            name='load_aggregated_skips_and_saves',
            tasks=base.SyncRunner(
                load_fact_tables_tasks_sf.load_aggregated_skips_and_saves.fill(
                    namespace='load_aggregated_skips_and_saves',
                    feed_name='bootstrap_feed.feed_name',
                    secrets_path=StaticParam(config.secrets_path),
                    date='bootstrap_feed.date',
                    sfdb_params='sfdb_params',
                    kwargs='bootstrap_feed.common_kwargs')))

    @property
    def load_fact_tables(self):
        """Activity to load fact_analytics & fact_analytics_error.

        Deletes any existing data for this feed and date in fact_analytics &
        fact_analytics_error.
        """
        return self.create(
            name='load_fact_tables',
            tasks=base.SyncRunner(
                load_fact_tables_tasks_sf.load_fact_data.fill(
                    namespace='load_fact_data',
                    feed_name='bootstrap_feed.feed_name',
                    secrets_path=StaticParam(config.secrets_path),
                    date='bootstrap_feed.date',
                    sfdb_params='sfdb_params',
                    kwargs='bootstrap_feed.common_kwargs'),
                overall_status_tasks.set_overall_status.fill(
                    feed_name='bootstrap_feed.feed_name',
                    date='bootstrap_feed.date',
                    set_status_once=StaticParam(True),
                    status=StaticParam(
                        garcon_feed_status.STATUS_INGESTED))))

    @property
    def clean_up(self):
        """Clean up tasks after flow is complete."""
        return self.create(
            name='clean_up',
            tasks=base.SyncRunner(
                tasks.remove_file_stage.fill(
                    namespace='remove_file_stage',
                    date='bootstrap_feed.date',
                    report_name='report_name',
                    licensor='bootstrap_feed.licensor')))

    @property
    def monitor_drop_location(self):
        """Check for files with new territories.

        Generate subject and message for SNS message if new territories
        are found.
        """
        return self.create(
            name='monitor_drop_location',
            tasks=base.SyncRunner(
                tasks.monitor_drop_location.fill(
                    namespace='monitor_drop_location',
                    s3_drop_location='bootstrap_feed.s3_drop_location',
                    countries='bootstrap_feed.countries',
                    report_name='report_name',
                    skip_monitor='skip_monitor',
                    date='bootstrap_feed.date',
                    licensor='bootstrap_feed.licensor')))

    @property
    def send_unexpected_files_sns_notification(self):
        """Send SNS message with unexpected files."""
        return self.create(
            name='send_unexpected_files_sns_notification',
            tasks=base.SyncRunner(
                garcon_sns.sns_publish_message.fill(
                    topic=StaticParam(self.sns_topic),
                    message='monitor_drop_location.unexpected_files_message',
                    subject='monitor_drop_location.unexpected_files_subject')))

    @property
    def send_stability_report_sns_notification(self):
        """Send SNS message with stability report."""
        return self.create(
            name='send_stability_report_sns_notification',
            tasks=base.SyncRunner(
                garcon_sns.sns_publish_message.fill(
                    topic=StaticParam(self.sns_topic),
                    message='monitor_drop_location.stability_files_message',
                    subject='monitor_drop_location.stability_files_subject')))

    @property
    def build_jenkins_dbt(self):
        """Build Jenkins DBT job."""
        return self.create(
            name='build_jenkins_dbt',
            schedule_to_start=48000,
            tasks=base.SyncRunner(
                build_jenkins_dbt.fill(
                    namespace='build_jenkins_dbt',
                    date='bootstrap_feed.date',
                    feed_name='bootstrap_feed.feed_name',
                    licensor='bootstrap_feed.licensor',
                    build_dbt='build_dbt',
                    # coming from context of the https://scheduler.theorchard.io/job/swf-amazon-music-exec/  # noqa
                    config='bootstrap_feed.jenkins_config')))

    def _is_update_feed_status_available(self, update_feed_s3_file_status):
        feed_status = update_feed_s3_file_status.result.get(
            'update_feed_s3_file_status.file_status')
        return feed_status == garcon_feed_status.STATUS_NOT_AVAILABLE