feat: fix events handler on ledger v2 (#538)

.github/workflows/main.yml

@@ -39,8 +39,8 @@ jobs:
           GITHUB_TOKEN: ${{ secrets.NUMARY_GITHUB_TOKEN }}
           SPEAKEASY_API_KEY: ${{ secrets.SPEAKEASY_API_KEY }}
           FURY_TOKEN: ${{ secrets.FURY_TOKEN }}
-          SEGMENT_WRITE_KEY: ${{ secrets.SEGMENT_WRITE_KEY_OSS }}
           GORELEASER_KEY: ${{ secrets.GORELEASER_KEY }}
+
   GoReleaser:
     runs-on: "formance-runner"
     if: contains(github.event.pull_request.labels.*.name, 'build-images') || github.ref == 'refs/heads/main'

Earthfile

@@ -34,8 +34,8 @@ build-final-spec:
     WORKDIR /src/openapi
     COPY openapi/openapi-merge.json .
     RUN npm run build
-    ENV VERSION v1.0.$(date +%Y%m%d)
-    RUN jq '.info.version = "${VERSION}"' build/generate.json > build/generate-with-version.json
+    LET VERSION=$(date +%Y%m%d)
+    RUN jq '.info.version = "v1.0.${VERSION}"' build/generate.json > build/generate-with-version.json
     SAVE ARTIFACT build/generate-with-version.json
     SAVE ARTIFACT build/generate-with-version.json AS LOCAL openapi/build/generate.json
 
@@ -182,7 +182,10 @@ pre-commit:
     FOR component IN $(ls components)
         BUILD --pass-args ./components/$component+pre-commit
     END
-    BUILD --pass-args +integration-tests
+    ARG skipIntegrationTests=0
+    IF [ "$skipIntegrationTests" = "0" ]
+        BUILD --pass-args +integration-tests
+    END
 
 pr:
     LOCALLY

components/ledger/internal/storage/ledgerstore/logs_test.go

@@ -3,11 +3,12 @@ package ledgerstore
 import (
 	"context"
 	"fmt"
-	"github.com/formancehq/stack/libs/go-libs/logging"
 	"math/big"
 	"testing"
 	"time"
 
+	"github.com/formancehq/stack/libs/go-libs/logging"
+
 	"github.com/formancehq/ledger/internal/storage/sqlutils"
 
 	ledger "github.com/formancehq/ledger/internal"

components/ledger/internal/storage/ledgerstore/store_benchmarks_test.go

@@ -4,15 +4,16 @@ import (
 	"context"
 	"flag"
 	"fmt"
+	"math/big"
+	"testing"
+
 	ledger "github.com/formancehq/ledger/internal"
 	"github.com/formancehq/stack/libs/go-libs/bun/bunexplain"
 	"github.com/formancehq/stack/libs/go-libs/logging"
 	"github.com/formancehq/stack/libs/go-libs/metadata"
 	"github.com/formancehq/stack/libs/go-libs/query"
 	"github.com/stretchr/testify/require"
 	"github.com/uptrace/bun"
-	"math/big"
-	"testing"
 )
 
 var nbTransactions = flag.Int("transactions", 10000, "number of transactions to create")

components/ledger/internal/storage/ledgerstore/transactions.go

@@ -6,15 +6,16 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"math/big"
+	"regexp"
+	"strings"
+
 	ledger "github.com/formancehq/ledger/internal"
 	"github.com/formancehq/ledger/internal/storage/paginate"
 	"github.com/formancehq/stack/libs/go-libs/api"
 	"github.com/formancehq/stack/libs/go-libs/metadata"
 	"github.com/formancehq/stack/libs/go-libs/query"
 	"github.com/uptrace/bun"
-	"math/big"
-	"regexp"
-	"strings"
 )
 
 const (

components/operator/internal/controllers/stack/testdata/monopod-disabled-one-service/results/configmaps--v1/search-benthos-resources.yaml

@@ -1,22 +1,33 @@
 apiVersion: v1
 data:
-    output_elasticsearch.yaml: |
+    output_elasticsearch.yaml: |-
         output_resources:
         - label: elasticsearch
           processors:
           - mapping: |
-              root = this.merge({
-                "stack": env("STACK")
-              })
-              meta id = "%s-%s".format(env("STACK"), meta("id"))
+              root = match {
+                this.action == "update" => if this.exists("upsert") { this.merge({
+                  "upsert": {
+                    "stack": env("STACK")
+                  }
+                }) } else { this },
+                _ => this.merge({
+                  "document": {
+                    "stack": env("STACK")
+                  }
+                })
+              }
+              root.id = "%s-%s".format(env("STACK"), this.id)
+          - log:
+              message: "Will write: ${! this }"
           elasticsearch:
             urls:
             - ${OPENSEARCH_URL}
-            id: ${!meta("id")}
+            id: ${! json().id }
             sniff: false
             healthcheck: false
             index: ${OPENSEARCH_INDEX}
-            action: ${!meta("action")}
+            action: ${! json().action }
             tls:
               enabled: true
               skip_cert_verify: true
@@ -25,8 +36,12 @@ data:
               username: ${BASIC_AUTH_USERNAME}
               password: ${BASIC_AUTH_PASSWORD}
             batching:
-              count: ${OPENSEARCH_BATCHING_COUNT}
-              period: ${OPENSEARCH_BATCHING_PERIOD}
+              count: ${OPENSEARCH_BATCHING_COUNT:10}
+              period: ${OPENSEARCH_BATCHING_PERIOD:1s}
+            multipart: true
+            doc: ${! if json().exists("document") { json("document") } else {""} }
+            script: ${! if json().exists("script") { json("script") } else {""} }
+            upsert: ${! if json().exists("upsert") { json("upsert") } else {""} }
 kind: ConfigMap
 metadata:
     labels:

components/operator/internal/controllers/stack/testdata/monopod-disabled-one-service/results/configmaps--v1/search-benthos-templates.yaml

@@ -36,40 +36,6 @@ data:
               "durable": this.consumer_group
             }
           }}
-    get_doc.yaml: |
-        name: get_doc
-        type: processor
-
-        fields:
-        - name: id
-          type: string
-
-        mapping: |
-          root = {
-            "branch": {
-              "request_map": "root = \"\"",
-              "result_map": "root._doc = this",
-              "processors": [
-                {
-                  "http": {
-                    "url": "%s/%s/_doc/%s-%s".format(env("OPENSEARCH_URL"), env("OPENSEARCH_INDEX"), env("STACK"), this.id),
-                    "verb": "GET",
-                    "basic_auth": {
-                      "enabled": env("BASIC_AUTH_ENABLED") == "true",
-                      "username": env("BASIC_AUTH_USERNAME"),
-                      "password": env("BASIC_AUTH_PASSWORD")
-                    },
-                    "tls": {
-                      "enabled": true,
-                      "skip_cert_verify": true
-                    }
-                  }
-                }, {
-                  "bloblang": "root = this._source"
-                }
-              ]
-            }
-          }
     postgres_query.yaml: |
         name: postgres_query
         type: processor

components/operator/internal/controllers/stack/testdata/monopod-disabled-one-service/results/deployments-apps-v1/monopod-disabled-one-service.yaml

@@ -527,8 +527,8 @@ spec:
                       value: $(LEDGER_POSTGRES_NO_DATABASE_URI)/$(LEDGER_POSTGRES_DATABASE)?sslmode=disable
                     - name: BROKER
                       value: kafka
-                  image: public.ecr.aws/h9j1u6h3/jeffail/benthos:4.12.1
-                  imagePullPolicy: Always
+                  image: public.ecr.aws/formance-internal/jeffail/benthos:v4.23.0-es
+                  imagePullPolicy: IfNotPresent
                   name: search-benthos
                   ports:
                     - containerPort: 4195

components/operator/internal/controllers/stack/testdata/monopod-latest/results/configmaps--v1/search-benthos-resources.yaml

@@ -1,22 +1,33 @@
 apiVersion: v1
 data:
-    output_elasticsearch.yaml: |
+    output_elasticsearch.yaml: |-
         output_resources:
         - label: elasticsearch
           processors:
           - mapping: |
-              root = this.merge({
-                "stack": env("STACK")
-              })
-              meta id = "%s-%s".format(env("STACK"), meta("id"))
+              root = match {
+                this.action == "update" => if this.exists("upsert") { this.merge({
+                  "upsert": {
+                    "stack": env("STACK")
+                  }
+                }) } else { this },
+                _ => this.merge({
+                  "document": {
+                    "stack": env("STACK")
+                  }
+                })
+              }
+              root.id = "%s-%s".format(env("STACK"), this.id)
+          - log:
+              message: "Will write: ${! this }"
           elasticsearch:
             urls:
             - ${OPENSEARCH_URL}
-            id: ${!meta("id")}
+            id: ${! json().id }
             sniff: false
             healthcheck: false
             index: ${OPENSEARCH_INDEX}
-            action: ${!meta("action")}
+            action: ${! json().action }
             tls:
               enabled: true
               skip_cert_verify: true
@@ -25,8 +36,12 @@ data:
               username: ${BASIC_AUTH_USERNAME}
               password: ${BASIC_AUTH_PASSWORD}
             batching:
-              count: ${OPENSEARCH_BATCHING_COUNT}
-              period: ${OPENSEARCH_BATCHING_PERIOD}
+              count: ${OPENSEARCH_BATCHING_COUNT:10}
+              period: ${OPENSEARCH_BATCHING_PERIOD:1s}
+            multipart: true
+            doc: ${! if json().exists("document") { json("document") } else {""} }
+            script: ${! if json().exists("script") { json("script") } else {""} }
+            upsert: ${! if json().exists("upsert") { json("upsert") } else {""} }
 kind: ConfigMap
 metadata:
     labels: