Enable setting the backfill counter of all task bindings with a single action

src/components/capture/AutoDiscoverySettings.tsx

@@ -142,8 +142,8 @@ function AutoDiscoverySettings({ readOnly }: Props) {
     ]);
 
     return (
-        <Stack spacing={1} sx={{ mt: 2, mb: 3 }}>
-            <Typography sx={{ fontWeight: 500 }}>
+        <Stack spacing={1}>
+            <Typography variant="formSectionHeader">
                 <FormattedMessage id="workflows.autoDiscovery.header" />
             </Typography>
 

src/components/collection/ResourceConfig.tsx

@@ -1,7 +1,7 @@
 import { Box, Typography } from '@mui/material';
 import ResourceConfigForm from 'components/collection/ResourceConfigForm';
+import Backfill from 'components/editor/Bindings/Backfill';
 import FieldSelectionViewer from 'components/editor/Bindings/FieldSelection';
-import ManualBackfill from 'components/editor/Bindings/ManualBackfill';
 import TimeTravel from 'components/editor/Bindings/TimeTravel';
 import { useEditorStore_queryResponse_draftedBindingIndex } from 'components/editor/Store/hooks';
 import { useEntityType } from 'context/EntityContext';
@@ -54,7 +54,14 @@ function ResourceConfig({ collectionName, readOnly = false }: Props) {
             </Box>
 
             {isEdit && draftedBindingIndex > -1 && !collectionDisabled ? (
-                <ManualBackfill bindingIndex={draftedBindingIndex} />
+                <Backfill
+                    bindingIndex={draftedBindingIndex}
+                    description={
+                        <FormattedMessage
+                            id={`workflows.collectionSelector.manualBackfill.message.${entityType}`}
+                        />
+                    }
+                />
             ) : null}
 
             {entityType === 'materialization' && !collectionDisabled ? (

src/components/editor/Bindings/Backfill/index.tsx

@@ -0,0 +1,189 @@
+import { Box, Stack, Typography } from '@mui/material';
+import OutlinedToggleButton from 'components/shared/OutlinedToggleButton';
+import { Check } from 'iconoir-react';
+import { ReactNode, useCallback, useMemo } from 'react';
+import { FormattedMessage } from 'react-intl';
+import {
+    useFormStateStore_isActive,
+    useFormStateStore_setFormState,
+} from 'stores/FormState/hooks';
+import { FormStatus } from 'stores/FormState/types';
+import {
+    useResourceConfig_allBindingsDisabled,
+    useResourceConfig_backfillAllBindings,
+    useResourceConfig_backfilledCollections,
+    useResourceConfig_collections,
+    useResourceConfig_currentCollection,
+    useResourceConfig_setBackfilledCollections,
+} from 'stores/ResourceConfig/hooks';
+import { hasLength } from 'utils/misc-utils';
+import { useEditorStore_queryResponse_draftSpecs } from '../../Store/hooks';
+import useUpdateBackfillCounter, {
+    BindingMetadata,
+} from './useUpdateBackfillCounter';
+
+export type BooleanString = 'true' | 'false';
+
+interface Props {
+    description: ReactNode;
+    bindingIndex?: number;
+}
+
+function Backfill({ description, bindingIndex = -1 }: Props) {
+    const { updateBackfillCounter } = useUpdateBackfillCounter();
+
+    // Draft Editor Store
+    const draftSpecs = useEditorStore_queryResponse_draftSpecs();
+
+    // Form State Store
+    const formActive = useFormStateStore_isActive();
+    const setFormState = useFormStateStore_setFormState();
+
+    // Resource Config Store
+    const currentCollection = useResourceConfig_currentCollection();
+    const collections = useResourceConfig_collections();
+
+    const backfilledCollections = useResourceConfig_backfilledCollections();
+    const setBackfilledCollections =
+        useResourceConfig_setBackfilledCollections();
+
+    const backfillAllBindings = useResourceConfig_backfillAllBindings();
+    const allBindingsDisabled = useResourceConfig_allBindingsDisabled();
+
+    const selected = useMemo(() => {
+        if (bindingIndex === -1) {
+            return backfillAllBindings;
+        }
+
+        return currentCollection
+            ? backfilledCollections.includes(currentCollection)
+            : false;
+    }, [
+        backfillAllBindings,
+        backfilledCollections,
+        bindingIndex,
+        currentCollection,
+    ]);
+
+    const value: BooleanString = useMemo(
+        () => (selected ? 'true' : 'false'),
+        [selected]
+    );
+
+    const draftSpec = useMemo(
+        () =>
+            draftSpecs.length > 0 && draftSpecs[0].spec ? draftSpecs[0] : null,
+        [draftSpecs]
+    );
+
+    const evaluateServerDifferences = useCallback(
+        (increment: BooleanString) => {
+            if (bindingIndex === -1) {
+                return (
+                    (backfillAllBindings && increment === 'false') ||
+                    (!backfillAllBindings && increment === 'true')
+                );
+            }
+
+            if (currentCollection) {
+                return increment === 'true'
+                    ? !backfilledCollections.includes(currentCollection)
+                    : backfilledCollections.includes(currentCollection);
+            }
+
+            return false;
+        },
+        [
+            backfillAllBindings,
+            backfilledCollections,
+            bindingIndex,
+            currentCollection,
+        ]
+    );
+
+    const handleClick = useCallback(
+        (increment: BooleanString) => {
+            const serverUpdateRequired = evaluateServerDifferences(increment);
+
+            if (draftSpec && serverUpdateRequired) {
+                setFormState({ status: FormStatus.UPDATING, error: null });
+
+                const singleBindingUpdate =
+                    bindingIndex > -1 && currentCollection;
+
+                const bindingMetadata: BindingMetadata[] = singleBindingUpdate
+                    ? [{ collection: currentCollection, bindingIndex }]
+                    : [];
+
+                updateBackfillCounter(
+                    draftSpec,
+                    increment,
+                    bindingMetadata
+                ).then(
+                    () => {
+                        const targetCollection = singleBindingUpdate
+                            ? currentCollection
+                            : undefined;
+
+                        setBackfilledCollections(increment, targetCollection);
+                        setFormState({ status: FormStatus.UPDATED });
+                    },
+                    (error) => {
+                        setFormState({
+                            status: FormStatus.FAILED,
+                            error: {
+                                title: 'workflows.collectionSelector.manualBackfill.error.title',
+                                error,
+                            },
+                        });
+                    }
+                );
+            }
+        },
+        [
+            bindingIndex,
+            currentCollection,
+            draftSpec,
+            evaluateServerDifferences,
+            setBackfilledCollections,
+            setFormState,
+            updateBackfillCounter,
+        ]
+    );
+
+    return (
+        <Box sx={{ mt: 3 }}>
+            <Stack spacing={1} sx={{ mb: 2 }}>
+                <Typography
+                    variant={bindingIndex === -1 ? 'formSectionHeader' : 'h6'}
+                >
+                    <FormattedMessage id="workflows.collectionSelector.manualBackfill.header" />
+                </Typography>
+
+                <Typography component="div">{description}</Typography>
+            </Stack>
+
+            <OutlinedToggleButton
+                value={value}
+                selected={selected}
+                disabled={
+                    formActive || !hasLength(collections) || allBindingsDisabled
+                }
+                onClick={(event, checked: string) => {
+                    event.preventDefault();
+                    event.stopPropagation();
+
+                    handleClick(checked === 'true' ? 'false' : 'true');
+                }}
+            >
+                <FormattedMessage id="workflows.collectionSelector.manualBackfill.cta.backfill" />
+
+                {selected ? (
+                    <Check style={{ marginLeft: 8, fontSize: 13 }} />
+                ) : null}
+            </OutlinedToggleButton>
+        </Box>
+    );
+}
+
+export default Backfill;

src/components/editor/Bindings/Backfill/useUpdateBackfillCounter.ts

@@ -0,0 +1,134 @@
+import { modifyDraftSpec } from 'api/draftSpecs';
+import { BooleanString } from 'components/editor/Bindings/Backfill';
+import { useEntityType } from 'context/EntityContext';
+import { DraftSpecQuery } from 'hooks/useDraftSpecs';
+import { useCallback } from 'react';
+import { useIntl } from 'react-intl';
+import { BASE_ERROR } from 'services/supabase';
+import { useResourceConfig_backfilledCollections } from 'stores/ResourceConfig/hooks';
+import { Schema } from 'types';
+import { hasLength } from 'utils/misc-utils';
+import { getBackfillCounter, getCollectionName } from 'utils/workflow-utils';
+import {
+    useEditorStore_persistedDraftId,
+    useEditorStore_queryResponse_mutate,
+} from '../../Store/hooks';
+
+export interface BindingMetadata {
+    bindingIndex: number;
+    collection: string;
+}
+
+const evaluateBackfillCounter = (
+    binding: Schema,
+    increment: BooleanString
+): number => {
+    let counter = getBackfillCounter(binding);
+
+    if (increment === 'true') {
+        counter = counter + 1;
+    } else if (counter > 0) {
+        counter = counter - 1;
+    }
+
+    return counter;
+};
+
+function useUpdateBackfillCounter() {
+    const intl = useIntl();
+    const entityType = useEntityType();
+
+    // Draft Editor Store
+    const draftId = useEditorStore_persistedDraftId();
+    const mutateDraftSpecs = useEditorStore_queryResponse_mutate();
+
+    // Resource Config Store
+    const backfilledCollections = useResourceConfig_backfilledCollections();
+
+    const updateBackfillCounter = useCallback(
+        async (
+            draftSpec: DraftSpecQuery,
+            increment: BooleanString,
+            bindingMetadata: BindingMetadata[]
+        ) => {
+            const bindingMetadataExists = hasLength(bindingMetadata);
+
+            const invalidBindingIndex = bindingMetadataExists
+                ? bindingMetadata.findIndex(
+                      ({ bindingIndex }) => bindingIndex === -1
+                  )
+                : -1;
+
+            if (!mutateDraftSpecs || invalidBindingIndex > -1) {
+                const errorMessageId = bindingMetadataExists
+                    ? 'workflows.collectionSelector.manualBackfill.error.message.singleCollection'
+                    : 'workflows.collectionSelector.manualBackfill.error.message.allBindings';
+
+                const errorMessageValues = bindingMetadataExists
+                    ? {
+                          collection:
+                              bindingMetadata[invalidBindingIndex].collection,
+                      }
+                    : undefined;
+
+                return Promise.reject({
+                    ...BASE_ERROR,
+                    message: intl.formatMessage(
+                        { id: errorMessageId },
+                        errorMessageValues
+                    ),
+                });
+            }
+
+            const spec: Schema = draftSpec.spec;
+
+            if (bindingMetadataExists) {
+                bindingMetadata.forEach(({ bindingIndex }) => {
+                    if (bindingIndex > -1) {
+                        spec.bindings[bindingIndex].backfill =
+                            evaluateBackfillCounter(
+                                spec.bindings[bindingIndex],
+                                increment
+                            );
+                    }
+                });
+            } else {
+                spec.bindings.forEach((binding: Schema, index: number) => {
+                    const collection = getCollectionName(binding);
+                    const collectionBackfilled =
+                        backfilledCollections.includes(collection);
+
+                    const shouldIncrement =
+                        !collectionBackfilled && increment === 'true';
+
+                    const shouldDecrement =
+                        collectionBackfilled && increment === 'false';
+
+                    if (shouldIncrement || shouldDecrement) {
+                        spec.bindings[index].backfill = evaluateBackfillCounter(
+                            binding,
+                            increment
+                        );
+                    }
+                });
+            }
+
+            const updateResponse = await modifyDraftSpec(spec, {
+                draft_id: draftId,
+                catalog_name: draftSpec.catalog_name,
+                spec_type: entityType,
+            });
+
+            if (updateResponse.error) {
+                return Promise.reject(updateResponse.error);
+            }
+
+            return mutateDraftSpecs();
+        },
+        [backfilledCollections, draftId, entityType, intl, mutateDraftSpecs]
+    );
+
+    return { updateBackfillCounter };
+}
+
+export default useUpdateBackfillCounter;