Repository: kestra-io/kestra
Issue: Trigger files from internal storage cannot be retrieved from the Execution Overview
Labels: bug

### Expected Behavior

On the execution Overview tab, if a triggre produces a file there is a link to the file on the tab that should works.

### Actual Behaviour

The link didn't works

### Steps To Reproduce

_No response_

### Environment Information

- Kestra Version: 0.9.0-SNAPSHOT
- Operating System (OS / Docker / Kubernetes):
- Java Version (If not docker):


### Example flow

```
id: trigger-wikipedia
namespace: io.kestra.tests

triggers:
  - id: wikipedia
    type: io.kestra.plugin.gcp.bigquery.Trigger
    sql: |
      SELECT DATETIME(datehour) as date, title, views FROM `bigquery-public-data.wikipedia.pageviews_2023` 
      WHERE DATE(datehour) = current_date() and wiki = 'fr' and title not in ('Cookie_(informatique)', 'Wikipédia:Accueil_principal', 'Spécial:Recherche')
      ORDER BY datehour desc, views desc
      LIMIT 10
    store: true
    interval: PT1M

tasks:
  - id: write-csv
    type: io.kestra.plugin.serdes.csv.CsvWriter
    from: "{{trigger.uri}}"


```

Issue comments:
This is caused by the BigQuery trigger re-using the Query task.
In this case, the storage files are created without execution prefix like:
```
kestra:///io/kestra/tests/trigger-wikipedia/10807484584123982691.ion
```

A correct file created by a trigger would be like:
```
kestra:///io/kestra/tests/gcs-listen/executions/4WxaKPGmyHzQcTocTwrafP/trigger/watch/5816538197308626077.tmp
```

Which would be accepted by the `ExecutionController.validateFile()` methods that otherwise return a 422 status code.
@tchiotludo what will be the best way to fix that issue ?

We can allow accessing files that didn't contain `/executions/{executionId}/trigger/{triggerName}`, but maybe it's a security risk, or we can update the BigQuery trigger (and maybe ten other that should have the same issue) to create a file with a proper name (it may be hard as Trigger use the Query task so we loose the trigger context).
We can allow acces on to `/executions/{executionId}/trigger/{triggerName}` but we need to control the flow behind to if user is allow to download

Pull request review comments:
use IdUtils.create() 

Patch that resolved the issue:
diff --git a/core/src/main/java/io/kestra/core/runners/RunContext.java b/core/src/main/java/io/kestra/core/runners/RunContext.java
index bf93486822..c94e8c2e54 100644
--- a/core/src/main/java/io/kestra/core/runners/RunContext.java
+++ b/core/src/main/java/io/kestra/core/runners/RunContext.java
@@ -51,6 +51,8 @@ public class RunContext {
 
     protected transient Path temporaryDirectory;
 
+    private String triggerExecutionId;
+
     /**
      * Only used by {@link io.kestra.core.models.triggers.types.Flow}
      *
@@ -87,7 +89,8 @@ public RunContext(ApplicationContext applicationContext, Flow flow, Task task, E
     public RunContext(ApplicationContext applicationContext, Flow flow, AbstractTrigger trigger) {
         this.initBean(applicationContext);
 
-        this.storageOutputPrefix = this.storageInterface.outputPrefix(flow);
+        this.triggerExecutionId = IdUtils.create();
+        this.storageOutputPrefix = this.storageInterface.outputPrefix(flow, trigger, triggerExecutionId);
         this.variables = this.variables(flow, null, null, null, trigger);
         this.initLogger(flow, trigger);
     }
@@ -162,6 +165,10 @@ private void initLogger(Flow flow, AbstractTrigger trigger) {
         );
     }
 
+    public String getTriggerExecutionId() {
+        return triggerExecutionId;
+    }
+
     public Map<String, Object> getVariables() {
         return variables;
     }
diff --git a/core/src/main/java/io/kestra/core/storages/StorageInterface.java b/core/src/main/java/io/kestra/core/storages/StorageInterface.java
index d7240dbe36..e81fcce7b0 100644
--- a/core/src/main/java/io/kestra/core/storages/StorageInterface.java
+++ b/core/src/main/java/io/kestra/core/storages/StorageInterface.java
@@ -8,6 +8,7 @@
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.Input;
 import io.kestra.core.models.tasks.Task;
+import io.kestra.core.models.triggers.AbstractTrigger;
 import io.kestra.core.utils.Slugify;
 import io.micronaut.core.annotation.Introspected;
 import io.micronaut.core.annotation.Nullable;
@@ -158,7 +159,7 @@ default URI outputPrefix(Flow flow)  {
         }
     }
 
-    default URI outputPrefix(Flow flow, Task task, Execution execution, TaskRun taskRun)  {
+    default URI outputPrefix(Flow flow, Task task, Execution execution, TaskRun taskRun) {
         try {
             return new URI("/" + String.join(
                 "/",
@@ -176,4 +177,22 @@ default URI outputPrefix(Flow flow, Task task, Execution execution, TaskRun task
             throw new IllegalArgumentException(e);
         }
     }
+
+    default URI outputPrefix(Flow flow, AbstractTrigger trigger, String triggerExecutionId) {
+        try {
+            return new URI("/" + String.join(
+                "/",
+                Arrays.asList(
+                    flow.getNamespace().replace(".", "/"),
+                    Slugify.of(flow.getId()),
+                    "executions",
+                    triggerExecutionId,
+                    "trigger",
+                    Slugify.of(trigger.getId())
+                )
+            ));
+        } catch (URISyntaxException e) {
+            throw new IllegalArgumentException(e);
+        }
+    }
 }


Test patch: