Repository: kestra-io/kestra
Issue: Add a core Toggle task allowing to disable or reenable a Trigger
Labels: enhancement

### Feature description

EDIT: the API endpoint has already been added and https://github.com/kestra-io/kestra/issues/2912 will handle the UI part of adding the toggle UI button. This issue is to add the Toggle task. The  only remaining issue is figuring out EE permission for that task — we may need to require a service account property that specifically allows accessing triggers from the specified namespace.

Imagine the following flow:

```yaml
id: http
namespace: dev

tasks:
  - id: slack
    type: io.kestra.plugin.notifications.slack.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {
        "channel": "#general",
        "text": "The price is now: {{ json(trigger.body).price }}"
      }

triggers:
  - id: http
    type: io.kestra.plugin.fs.http.Trigger
    uri: https://fakestoreapi.com/products/1
    responseCondition: "{{ json(response.body).price <= 110 }}"
    interval: PT30S
```

This flow checks every 30 seconds if the price of an item drops below a certain price point. Once it does, it will send a Slack alert and disable the trigger (_We no longer need to poll for the status, as the condition is satisfied!_)

It would be great to add a POST API endpoint `/api/v1/triggers/toggle/by-query` and a task with equivalent functionality e.g.:

```yaml
      - id: disable_schedule
        type: io.kestra.core.tasks.triggers.Toggle
        namespace: dev
        flowId: http
        triggerId: http
        enabled: false # true to reenable
```


Full flow example for the same scenario:

```yaml
id: http
namespace: dev

tasks:
  - id: slack
    type: io.kestra.plugin.notifications.slack.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {
        "channel": "#general",
        "text": "The price is now: {{ json(trigger.body).price }}"
      }
  
  - id: if
    type: io.kestra.core.tasks.flows.If
    condition: "{{ json(trigger.body).price <= 110 }}"
    then:
      - id: disable_schedule
        type: io.kestra.core.tasks.triggers.Toggle
        namespace: dev
        flowId: http
        triggerId: http
        enabled: false # true to reenable

triggers:
  - id: http
    type: io.kestra.plugin.fs.http.Trigger
    uri: https://fakestoreapi.com/products/1
    responseCondition: "{{ json(response.body).price <= 110 }}"
    interval: PT30S
```

Once the condition is satisfied, the trigger is disabled.

## Alternative

An alternative might be in this scenario to stop polling in the HTTP trigger once the condition is satisfied.

Issue comments:
this might be a quicker workaround once the state store gets added as a condition:

```yaml
id: http
namespace: dev

tasks:
  - id: slack
    type: io.kestra.plugin.notifications.slack.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {
        "channel": "#general",
        "text": "The price is now: {{ json(trigger.body).price }}"
      }
  
  - id: if
    type: io.kestra.core.tasks.flows.If
    condition: "{{ json(trigger.body).price <= 110 }}"
    then:
      - id: setState
        type: io.kestra.core.tasks.states.Set
        data:
          disabled: true

triggers:
  - id: http
    type: io.kestra.plugin.fs.http.Trigger
    uri: https://fakestoreapi.com/products/1
    responseCondition: "{{ json(response.body).price <= 110 }}"
    interval: PT30S
    - conditions:
      - type: io.kestra.core.models.conditions.types.state.Get
        condition: "{{ disabled != true }}"
```
alternative option (less favorable but solves the immediate pain): at first, only provide the functionality within this task:

```yaml
id: http
namespace: dev

tasks:
  - id: slack
    type: io.kestra.plugin.notifications.slack.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {
        "channel": "#general",
        "text": "The price is now: {{ json(trigger.body).price }}"
      }

triggers:
  - id: http
    type: io.kestra.plugin.fs.http.Trigger
    uri: https://fakestoreapi.com/products/1
    responseCondition: "{{ json(response.body).price <= 110 }}"
    interval: PT30S
    stopAfterSuccess: true # false by default 
```
Next steps:
1. stopAfterSuccess boolean
2. Update trigger endpoint
3. UI change with toggle, unlock and Backfill Executions
4. Reevaluate after user feedback if we need the Toggle task and state store to store previous trigger conditions

to add more info from today's meeting with @brian-mulier-p and @tchiotludo: 
- we added an issue here https://github.com/kestra-io/kestra/issues/2964 for a generic `stopAfter` success property for all triggers
- we want to add the `Toggle` task allowing to disable a specific trigger (as shown below) via an API call (i.e. without changing the source code)
- we want to support more advanced use cases based on Flow trigger e.g. if a specific critical flow (or if any flow from a "prod" namespace) ends in a FAILED state, send a Slack alert + disable a trigger of that flow 
- for now, automatic reenabling of trigger is out of scope - user needs to reenable the trigger manually once they are ready

```yaml
      - id: disable_schedule
        type: io.kestra.core.tasks.triggers.Toggle
        namespace: dev
        flowId: http
        triggerId: http
        enabled: false # true to reenable
```
@anna-geller for permission, we can do the same as the Flow task: allowed to toggle the trigger if the current flow is allowed to trigger the target flow. If so, we can reuse the existing facility which would speed up development.
sounds good!

Pull request review comments:

Patch that resolved the issue:
diff --git a/core/src/main/java/io/kestra/core/tasks/trigger/Toggle.java b/core/src/main/java/io/kestra/core/tasks/trigger/Toggle.java
new file mode 100644
index 0000000000..be017f9906
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/tasks/trigger/Toggle.java
@@ -0,0 +1,138 @@
+package io.kestra.core.tasks.trigger;
+
+import io.kestra.core.models.annotations.Example;
+import io.kestra.core.models.annotations.Plugin;
+import io.kestra.core.models.annotations.PluginProperty;
+import io.kestra.core.models.tasks.RunnableTask;
+import io.kestra.core.models.tasks.Task;
+import io.kestra.core.models.tasks.VoidOutput;
+import io.kestra.core.models.triggers.Trigger;
+import io.kestra.core.models.triggers.TriggerContext;
+import io.kestra.core.queues.QueueFactoryInterface;
+import io.kestra.core.queues.QueueInterface;
+import io.kestra.core.repositories.TriggerRepositoryInterface;
+import io.kestra.core.runners.FlowExecutorInterface;
+import io.kestra.core.runners.RunContext;
+import io.micronaut.inject.qualifiers.Qualifiers;
+import io.swagger.v3.oas.annotations.media.Schema;
+import jakarta.validation.constraints.NotNull;
+import lombok.*;
+import lombok.experimental.SuperBuilder;
+
+import java.util.Map;
+import java.util.Optional;
+
+@SuperBuilder
+@ToString
+@EqualsAndHashCode
+@Getter
+@NoArgsConstructor
+@Schema(
+    title = "Toggle a trigger: enable or disable it."
+)
+@Plugin(
+    examples = {
+        @Example(
+            title = "Toggle a trigger on flow input.",
+            full = true,
+            code = {
+                """
+                    id: trigger-toggle
+                        namespace: myteam
+
+                        inputs:
+                          - id: toggle
+                            type: BOOLEAN
+                            defaults: true
+
+                        tasks:
+                          - id: if
+                            type: io.kestra.core.tasks.flows.If
+                            condition: "{{inputs.toggle}}"
+                            then:
+                            - id: enable
+                              type: io.kestra.core.tasks.trigger.Toggle
+                              trigger: schedule
+                              enabled: true
+                            else:
+                            - id: disable
+                              type: io.kestra.core.tasks.trigger.Toggle
+                              trigger: schedule
+                              enabled: false
+                          - id: log
+                            type: io.kestra.core.tasks.log.Log
+                            message: Hello World
+
+                        triggers:
+                          - id: schedule
+                            type: io.kestra.core.models.triggers.types.Schedule
+                            cron: "* * * * *\""""
+            }
+        )
+    }
+)
+public class Toggle extends Task implements RunnableTask<VoidOutput> {
+    @Schema(
+        title = "The flow identifier of the trigger to toggle.",
+        description = "If not set, the current flow identifier will be used."
+    )
+    @PluginProperty(dynamic = true)
+    private String flowId;
+
+    @Schema(
+        title = "The namespace of the flow of the trigger to toggle.",
+        description = "If not set, the current flow namespace will be used."
+    )
+    @PluginProperty(dynamic = true)
+    private String namespace;
+
+    @Schema(title = "The identifier of the trigger to toggle.")
+    @NotNull
+    @PluginProperty(dynamic = true)
+    private String trigger;
+
+    @Schema(title = "Whether to enable or disable the trigger.")
+    @NotNull
+    @Builder.Default
+    @PluginProperty
+    private Boolean enabled = false;
+
+    @Override
+    public VoidOutput run(RunContext runContext) throws Exception {
+        Map<String, String> flowVariables = (Map<String, String>) runContext.getVariables().get("flow");
+        String realNamespace = namespace == null ? flowVariables.get("namespace") : runContext.render(namespace);
+        String realFlowId = flowId == null ? flowVariables.get("id") : runContext.render(flowId);
+        String realTrigger = runContext.render(trigger);
+
+        // verify that the target flow exists, and the current execution is authorized to access it
+        FlowExecutorInterface flowExecutor = runContext.getApplicationContext().getBean(FlowExecutorInterface.class);
+        flowExecutor.findByIdFromFlowTask(
+            runContext.tenantId(),
+            realNamespace,
+            realFlowId,
+            Optional.empty(),
+            runContext.tenantId(),
+            flowVariables.get("namespace"),
+            flowVariables.get("id")
+        )
+            .orElseThrow(() -> new IllegalArgumentException("Unable to find flow " + realNamespace + "." + realFlowId + ". Make sure the flow exists and the current execution is authorized to access it."));
+
+
+        // load the trigger from the database
+        TriggerContext triggerContext = TriggerContext.builder()
+            .tenantId(runContext.tenantId())
+            .namespace(realNamespace)
+            .flowId(realFlowId)
+            .triggerId(realTrigger)
+            .build();
+        TriggerRepositoryInterface triggerRepository = runContext.getApplicationContext().getBean(TriggerRepositoryInterface.class);
+        Trigger currentTrigger = triggerRepository.findLast(triggerContext).orElseThrow(() -> new IllegalArgumentException("Unable to find trigger " + realTrigger + " for the flow " + realNamespace + "." + realFlowId));
+        currentTrigger = currentTrigger.toBuilder().disabled(!enabled).build();
+
+        // update the trigger by emitting inside the queue
+        QueueInterface<Trigger> triggerQueue = runContext.getApplicationContext().getBean(QueueInterface.class, Qualifiers.byName(QueueFactoryInterface.TRIGGER_NAMED));
+        triggerQueue.emit(currentTrigger);
+
+        return null;
+    }
+}
diff --git a/repository-memory/src/main/java/io/kestra/repository/memory/MemoryTriggerRepository.java b/repository-memory/src/main/java/io/kestra/repository/memory/MemoryTriggerRepository.java
index 16ac786b10..fcff81df62 100644
--- a/repository-memory/src/main/java/io/kestra/repository/memory/MemoryTriggerRepository.java
+++ b/repository-memory/src/main/java/io/kestra/repository/memory/MemoryTriggerRepository.java
@@ -21,7 +21,9 @@ public class MemoryTriggerRepository implements TriggerRepositoryInterface {
 
     @Override
     public Optional<Trigger> findLast(TriggerContext trigger) {
-        throw new UnsupportedOperationException();
+        return triggers.stream()
+            .filter(t -> t.uid().equals(trigger.uid()))
+            .findFirst();
     }
 
     @Override


Test patch:
diff --git a/core/src/test/java/io/kestra/core/tasks/trigger/ToggleTest.java b/core/src/test/java/io/kestra/core/tasks/trigger/ToggleTest.java
new file mode 100644
index 0000000000..28026a40ae
--- /dev/null
+++ b/core/src/test/java/io/kestra/core/tasks/trigger/ToggleTest.java
@@ -0,0 +1,61 @@
+package io.kestra.core.tasks.trigger;
+
+import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.flows.State;
+import io.kestra.core.models.triggers.Trigger;
+import io.kestra.core.queues.QueueFactoryInterface;
+import io.kestra.core.queues.QueueInterface;
+import io.kestra.core.repositories.TriggerRepositoryInterface;
+import io.kestra.core.runners.AbstractMemoryRunnerTest;
+import jakarta.inject.Inject;
+import jakarta.inject.Named;
+import org.junit.jupiter.api.Test;
+
+import java.time.ZonedDateTime;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.*;
+
+class ToggleTest extends AbstractMemoryRunnerTest {
+    @Inject
+    private TriggerRepositoryInterface triggerRepository;
+
+    @Inject
+    @Named(QueueFactoryInterface.TRIGGER_NAMED)
+    private QueueInterface<Trigger> triggerQueue;
+
+    @Test
+    void toggle() throws Exception {
+        Trigger trigger = Trigger
+            .builder()
+            .triggerId("schedule")
+            .flowId("trigger-toggle")
+            .namespace("io.kestra.tests")
+            .date(ZonedDateTime.now())
+            .disabled(true)
+            .build();
+        triggerRepository.save(trigger);
+
+        AtomicReference<Trigger> triggerRef = new AtomicReference<>();
+        CountDownLatch countDownLatch = new CountDownLatch(1);
+        triggerQueue.receive(either -> {
+            if (either.isLeft()) {
+                triggerRef.set(either.getLeft());
+                countDownLatch.countDown();
+            }
+        });
+
+        Execution execution = runnerUtils.runOne(null, "io.kestra.tests", "trigger-toggle");
+
+        assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
+        assertThat(execution.getTaskRunList(), hasSize(1));
+
+        countDownLatch.await(10, TimeUnit.SECONDS);
+        assertThat(countDownLatch.getCount(), is(0L));
+        assertThat(triggerRef.get(), notNullValue());
+        assertThat(triggerRef.get().getDisabled(), is(false));
+    }
+}
\ No newline at end of file
diff --git a/core/src/test/resources/flows/valids/trigger-toggle.yaml b/core/src/test/resources/flows/valids/trigger-toggle.yaml
new file mode 100644
index 0000000000..78511cfa8e
--- /dev/null
+++ b/core/src/test/resources/flows/valids/trigger-toggle.yaml
@@ -0,0 +1,13 @@
+id: trigger-toggle
+namespace: io.kestra.tests
+
+tasks:
+    - id: disable
+      type: io.kestra.core.tasks.trigger.Toggle
+      trigger: schedule
+      enabled: true
+
+triggers:
+  - id: schedule
+    type: io.kestra.core.models.triggers.types.Schedule
+    cron: "* * * * *"
\ No newline at end of file