diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml
new file mode 100644
index 0000000..44140a7
--- /dev/null
+++ b/.github/workflows/build.yml
@@ -0,0 +1,89 @@
+name: Build
+on:
+  pull_request:
+    branches: ["**"]
+  push:
+    branches:
+      - main
+
+jobs:
+  install-tools:
+    if: "!contains(github.event.head_commit.message, 'Merge pull request')"
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-go@v5
+        with:
+          go-version-file: 'go.mod'
+      - name: Install all tools
+        run: make tools
+      - uses: actions/upload-artifact@v4
+        with:
+          name: tools
+          path: bin/
+
+  go-tests:
+    needs: install-tools
+    if: "!contains(github.event.head_commit.message, 'Merge pull request')"
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-go@v5
+        with:
+          go-version-file: 'go.mod'
+      - uses: actions/download-artifact@v4
+        with:
+          name: tools
+          path: bin/
+      - name: Set executable
+        run: chmod +x bin/*
+      - name: Run Go tests
+        run: make test-go
+
+  prometheus-alerts:
+    needs: install-tools
+    if: "!contains(github.event.head_commit.message, 'Merge pull request')"
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/download-artifact@v4
+        with:
+          name: tools
+          path: bin/
+      - name: Set executable
+        run: chmod +x bin/*
+      - name: Run Prometheus alert tests
+        run: make test-prometheus-alerts
+
+  prometheus-rules:
+    needs: install-tools
+    if: "!contains(github.event.head_commit.message, 'Merge pull request')"
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/download-artifact@v4
+        with:
+          name: tools
+          path: bin/
+      - name: Set executable
+        run: chmod +x bin/*
+      - name: Run Prometheus rules tests
+        run: make test-prometheus-rules
+  lint:
+    needs: install-tools
+    if: "!contains(github.event.head_commit.message, 'Merge pull request')"
+    runs-on: "ubuntu-latest"
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-go@v5
+        with:
+          go-version-file: 'go.mod'
+      - uses: actions/download-artifact@v4
+        with:
+          name: tools
+          path: bin/
+      - name: Set executable
+        run: chmod +x bin/*
+
+      - name: Run lint
+        run: make lint
\ No newline at end of file
diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml
deleted file mode 100644
index e5b620e..0000000
--- a/.github/workflows/lint.yml
+++ /dev/null
@@ -1,26 +0,0 @@
-name: golangci-lint
-on:
-  pull_request:
-    branches: ["**"]
-jobs:
-  golangci:
-    runs-on: "ubuntu-latest"
-
-    name: lint
-    steps:
-      - uses: actions/checkout@v4
-      - uses: actions/setup-go@v5
-        with:
-          go-version-file: 'go.mod'
-
-      - name: install golangci-lint
-        run: make tools-golangci-lint
-
-      - name: tidy
-        run: go mod tidy
-
-      - name: build
-        run: go build
-
-      - name: Run golangci-lint
-        run: make lint
\ No newline at end of file
diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml
deleted file mode 100644
index ef14f4b..0000000
--- a/.github/workflows/tests.yml
+++ /dev/null
@@ -1,28 +0,0 @@
-name: Tests
-on:
-  pull_request:
-    branches: ["**"]
-  push:
-    branches:
-      - main
-
-jobs:
-  rules-test:
-    runs-on: ubuntu-latest 
-
-    steps:
-      - name: Checkout code
-        uses: actions/checkout@v4
-
-      # using go to get arch and os
-      - name: Install Go
-        uses: actions/setup-go@v5
-        with:
-          go-version-file: 'go.mod'
-
-      # we need to install promtool to run the tests
-      - name: Install promtool
-        run: make tools
-
-      - name: Run tests
-        run: make test
diff --git a/Makefile b/Makefile
index 11e5e85..75ef6b6 100644
--- a/Makefile
+++ b/Makefile
@@ -53,9 +53,9 @@ dep:
 	@go mod tidy
 
 lint-benthos: build  ## Run Benthos linter
-	@CLICKHOUSE_HOST="" CLICKHOUSE_PORT="" CLICKHOUSE_SIGNAL_DATABASE="" CLICKHOUSE_INDEX_DATABASE=""  CLICKHOUSE_USER="" CLICKHOUSE_PASSWORD="" \
+	@CLICKHOUSE_HOST="" CLICKHOUSE_PORT="" CLICKHOUSE_DIMO_DATABASE="" CLICKHOUSE_INDEX_DATABASE=""  CLICKHOUSE_USER="" CLICKHOUSE_PASSWORD="" \
 		S3_AWS_ACCESS_KEY_ID="" S3_AWS_SECRET_ACCESS_KEY="" S3_CLOUDEVENT_BUCKET="" S3_EPHEMERAL_BUCKET="" S3_AWS_REGION="" \
-	dps lint -r ./charts/dps/files/resources.yaml ./charts/dps/files/config.yaml ./charts/dps/files/*
+	dps lint -r ./charts/dps/files/resources.yaml ./charts/dps/files/config.yaml ./charts/dps/files/streams/*
 
 lint: lint-benthos ## Run linter for benthos config and go code
 	golangci-lint version
@@ -69,10 +69,16 @@ test-go: ## Run Go tests
 
 test: test-go test-prom ## run all tests
 
-test-prom: ## run prometheus tests
-	sed "s/{{ .Release.Namespace }}/dev/g"  ./charts/dps/templates/alerts.yaml |  sed 's/{{.*}}//g' >  tests/prom/alerts-modified.yaml
-	go tool promtool check rules tests/prom/alerts-modified.yaml
-	go tool promtool test rules tests/prom/rules-tests.yaml
+
+test-prometheus-prepare: ## Prepare Prometheus alert files for testing
+	@mkdir -p ./charts/dps/tests
+	@sed "s/{{ .Release.Namespace }}/dev/g" ./charts/dps/templates/alerts.yaml | sed 's/{{.*}}//g' > ./tests/prom/alerts-modified.yaml
+
+test-prometheus-alerts: test-prometheus-prepare ## Check Prometheus alert rules
+	@go tool promtool check rules ./tests/prom/alerts-modified.yaml
+
+test-prometheus-rules: test-prometheus-prepare ## Run Prometheus rules tests
+	@go tool promtool test rules ./tests/prom/rules-tests.yaml
 
 tools-golangci-lint:
 	@mkdir -p $(PATHINSTBIN)
diff --git a/charts/dps/files/config.yaml b/charts/dps/files/config.yaml
index 7087a27..dc4b91e 100644
--- a/charts/dps/files/config.yaml
+++ b/charts/dps/files/config.yaml
@@ -6,7 +6,7 @@ logger:
   add_timestamp: true
   format: json
   static_fields:
-    'appname': zone.dimo.dps
+    'app': zone.dimo.dps
 
 metrics:
   prometheus: {}
diff --git a/charts/dps/files/streams/events.yaml b/charts/dps/files/streams/events.yaml
new file mode 100644
index 0000000..8862a08
--- /dev/null
+++ b/charts/dps/files/streams/events.yaml
@@ -0,0 +1,55 @@
+input:
+  label: kafka_input
+  kafka_franz:
+    seed_brokers:
+      - ${KAFKA_BOOTSTRAP_SERVERS:localhost}:${KAFKA_BOOTSTRAP_PORT:9092}
+    topics:
+      - '${DEVICE_EVENTS_TOPIC:topic.device.events}'
+    consumer_group: "zone.dimo.dps.events"
+    client_id: ${CONTAINER_NAME:localhost}-dps-events
+    rack_id: ${NODE_NAME:localhost}
+    fetch_max_bytes: 100MiB
+    fetch_max_wait: 50ms
+    fetch_min_bytes: 1B
+    fetch_max_partition_bytes: 10MiB
+    commit_period: 2s
+    checkpoint_limit: 1000000 # double the batch size for insertion
+    start_offset: "latest"
+
+pipeline:
+  processors:
+    - label: "convert_events_to_slice"
+      dimo_event_to_slice: {}
+
+output:
+  label: "insert_event"
+  fallback:
+    - label: "insert_event_clickhouse"
+      sql_insert:
+        driver: clickhouse
+        dsn: clickhouse://${CLICKHOUSE_HOST}:${CLICKHOUSE_PORT}/${CLICKHOUSE_DIMO_DATABASE}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&secure=true&max_execution_time=600
+        table: event
+        columns: []
+        args_mapping: root = this
+        batching:
+          count: 500000
+          byte_size: 0
+          period: "2s"
+          check: ""
+
+    - label: "insert_event_clickhouse_failure"
+      switch:
+        cases:
+          # If the error contains "bad connection", reject the message so it will be retried
+          - check: 'metadata("fallback_error").or("").contains("bad connection")'
+            output:
+              reject: metadata("fallback_error")
+              processors:
+                - resource: "handle_db_connection_error"
+          - check: ''
+            output:
+              drop: {}
+              processors:
+                -  mutation: |
+                    meta dimo_component = "insert_event_clickhouse"
+                - resource: "handle_db_error"
diff --git a/charts/dps/files/partial-cloudevents.yaml b/charts/dps/files/streams/partial-cloudevents.yaml
similarity index 100%
rename from charts/dps/files/partial-cloudevents.yaml
rename to charts/dps/files/streams/partial-cloudevents.yaml
diff --git a/charts/dps/files/signals.yaml b/charts/dps/files/streams/signals.yaml
similarity index 84%
rename from charts/dps/files/signals.yaml
rename to charts/dps/files/streams/signals.yaml
index 757b314..098b3cd 100644
--- a/charts/dps/files/signals.yaml
+++ b/charts/dps/files/streams/signals.yaml
@@ -19,8 +19,8 @@ input:
 pipeline:
   processors:
     - label: signal_db_migration
-      dimo_signal_migration:
-        dsn: clickhouse://${CLICKHOUSE_HOST}:${CLICKHOUSE_PORT}/${CLICKHOUSE_SIGNAL_DATABASE}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&secure=true&dial_timeout=5s
+      dimo_db_migration:
+        dsn: clickhouse://${CLICKHOUSE_HOST}:${CLICKHOUSE_PORT}/${CLICKHOUSE_DIMO_DATABASE}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&secure=true&dial_timeout=5s
 
     - label: "inputlogger"
       for_each:
@@ -29,8 +29,8 @@ pipeline:
             message: 'MessageReceived'
             fields_mapping: |
               root.payload = this
-    - label: "convert_to_slice"
-      signal_to_slice:
+    - label: "convert_signals_to_slice"
+      dimo_signal_to_slice: {}
 
 output:
   label: "insert_signal"
@@ -38,7 +38,7 @@ output:
     - label: "insert_signal_clickhouse"
       sql_insert:
         driver: clickhouse
-        dsn: clickhouse://${CLICKHOUSE_HOST}:${CLICKHOUSE_PORT}/${CLICKHOUSE_SIGNAL_DATABASE}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&secure=true&max_execution_time=600
+        dsn: clickhouse://${CLICKHOUSE_HOST}:${CLICKHOUSE_PORT}/${CLICKHOUSE_DIMO_DATABASE}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&secure=true&max_execution_time=600
         table: signal
         columns: []
         args_mapping: root = this
diff --git a/charts/dps/files/valid-cloudevents.yaml b/charts/dps/files/streams/valid-cloudevents.yaml
similarity index 99%
rename from charts/dps/files/valid-cloudevents.yaml
rename to charts/dps/files/streams/valid-cloudevents.yaml
index 79f0fa3..cf1a6b8 100644
--- a/charts/dps/files/valid-cloudevents.yaml
+++ b/charts/dps/files/streams/valid-cloudevents.yaml
@@ -20,7 +20,7 @@ input:
 pipeline:
   processors:
     - label: "file_index_migration"
-      dimo_file_index_migration:
+      dimo_file_index_db_migration:
         dsn: clickhouse://${CLICKHOUSE_HOST}:${CLICKHOUSE_PORT}/${CLICKHOUSE_INDEX_DATABASE}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&secure=true&dial_timeout=5s
 
     - label: "inputlogger"
diff --git a/charts/dps/templates/configmap.yaml b/charts/dps/templates/configmap.yaml
index 7625038..a014932 100644
--- a/charts/dps/templates/configmap.yaml
+++ b/charts/dps/templates/configmap.yaml
@@ -7,3 +7,5 @@ metadata:
 {{ include "dps.labels" . | indent 4 }}
 data:
 {{ tpl (.Files.Glob "files/*").AsConfig $ | indent 2 }}
+{{ tpl (.Files.Glob ("files/streams/*")).AsConfig $ | indent 2 }}
+
diff --git a/charts/dps/templates/deployment.yaml b/charts/dps/templates/deployment.yaml
index e547f12..ab5dab2 100644
--- a/charts/dps/templates/deployment.yaml
+++ b/charts/dps/templates/deployment.yaml
@@ -71,15 +71,12 @@ spec:
           resources:
 {{- toYaml .Values.resources | nindent 12 }}
           volumeMounts:
-            - name: signals
-              mountPath: /streams/signals.yaml
-              subPath: signals.yaml
-            - name: valid-cloudevents
-              mountPath: /streams/valid-cloudevents.yaml
-              subPath: valid-cloudevents.yaml
-            - name: partial-cloudevents
-              mountPath: /streams/partial-cloudevents.yaml
-              subPath: partial-cloudevents.yaml
+            {{- $files := .Files.Glob ("files/streams/*") }}
+            {{- range $path, $_ := $files }}
+            - name: {{ $path | base | replace "." "-" }}
+              mountPath: /streams/{{ $path | base }}
+              subPath: {{ $path | base }}
+            {{- end }}
             - name: config
               mountPath: /config
               subPath: config.yaml
diff --git a/charts/dps/values.yaml b/charts/dps/values.yaml
index 2edd3dc..114fd60 100644
--- a/charts/dps/values.yaml
+++ b/charts/dps/values.yaml
@@ -33,12 +33,13 @@ securityContext:
 env:
   KAFKA_BOOTSTRAP_SERVERS: kafka-dev-dimo-kafka-kafka-brokers
   DEVICE_SIGNALS_TOPIC: topic.device.signals
+  DEVICE_EVENTS_TOPIC: topic.device.events
   DEVICE_VALID_CE_TOPIC: topic.device.validcloudevents
   DEVICE_PARTIAL_CE_TOPIC: topic.device.partialcloudevents
   ELASTIC_MULTIPLE_NODE: false
   ENVIRONMENT: dev
   CLICKHOUSE_PORT: 9440
-  CLICKHOUSE_SIGNAL_DATABASE: dimo
+  CLICKHOUSE_DIMO_DATABASE: dimo
   CLICKHOUSE_INDEX_DATABASE: file_index
   S3_CLOUDEVENT_BUCKET: dimo-ingest-cloudevent-dev
   S3_EPHEMERAL_BUCKET: dimo-ingest-ephemeral-dev
@@ -113,6 +114,12 @@ kafka:
         compression.type: producer
         cleanup.policy: delete
         retention.ms: '7200000'
+    - name: topic.device.events
+      config:
+        segment.ms: '3600000'
+        compression.type: producer
+        cleanup.policy: delete
+        retention.ms: '7200000'
     - name: topic.device.validcloudevents
       config:
         segment.ms: '3600000'
diff --git a/go.mod b/go.mod
index 1cedaca..ebc647d 100644
--- a/go.mod
+++ b/go.mod
@@ -3,10 +3,10 @@ module github.com/DIMO-Network/dps
 go 1.24.2
 
 require (
-	github.com/ClickHouse/clickhouse-go/v2 v2.37.0
-	github.com/DIMO-Network/clickhouse-infra v0.0.3
+	github.com/ClickHouse/clickhouse-go/v2 v2.37.2
+	github.com/DIMO-Network/clickhouse-infra v0.0.4
 	github.com/DIMO-Network/cloudevent v0.1.2
-	github.com/DIMO-Network/model-garage v0.6.0
+	github.com/DIMO-Network/model-garage v0.6.5
 	github.com/redpanda-data/benthos/v4 v4.50.0
 	github.com/redpanda-data/connect/v4 v4.54.0
 	github.com/stretchr/testify v1.10.0
@@ -34,7 +34,7 @@ require (
 	github.com/Azure/azure-sdk-for-go/sdk/storage/azblob v1.6.0 // indirect
 	github.com/AzureAD/microsoft-authentication-library-for-go v1.3.3 // indirect
 	github.com/BurntSushi/toml v1.4.1-0.20240526193622-a339e1f7089c // indirect
-	github.com/ClickHouse/ch-go v0.66.0 // indirect
+	github.com/ClickHouse/ch-go v0.66.1 // indirect
 	github.com/Code-Hex/go-generics-cache v1.5.1 // indirect
 	github.com/GoogleCloudPlatform/grpc-gcp-go/grpcgcp v1.5.2 // indirect
 	github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.27.0 // indirect
@@ -49,7 +49,7 @@ require (
 	github.com/OneOfOne/xxhash v1.2.8 // indirect
 	github.com/alecthomas/kingpin/v2 v2.4.0 // indirect
 	github.com/alecthomas/units v0.0.0-20240927000941-0f3dac36c52b // indirect
-	github.com/andybalholm/brotli v1.1.1 // indirect
+	github.com/andybalholm/brotli v1.2.0 // indirect
 	github.com/apache/arrow-go/v18 v18.0.0 // indirect
 	github.com/armon/go-metrics v0.4.1 // indirect
 	github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2 // indirect
@@ -119,7 +119,7 @@ require (
 	github.com/emicklei/go-restful/v3 v3.11.0 // indirect
 	github.com/envoyproxy/go-control-plane/envoy v1.32.4 // indirect
 	github.com/envoyproxy/protoc-gen-validate v1.2.1 // indirect
-	github.com/ethereum/go-ethereum v1.15.11 // indirect
+	github.com/ethereum/go-ethereum v1.16.1 // indirect
 	github.com/facette/natsort v0.0.0-20181210072756-2cd4dd1e2dcb // indirect
 	github.com/fatih/color v1.18.0 // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect
@@ -128,7 +128,7 @@ require (
 	github.com/gabriel-vasile/mimetype v1.4.7 // indirect
 	github.com/go-faster/city v1.0.1 // indirect
 	github.com/go-faster/errors v0.7.1 // indirect
-	github.com/go-logr/logr v1.4.2 // indirect
+	github.com/go-logr/logr v1.4.3 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/go-openapi/analysis v0.23.0 // indirect
 	github.com/go-openapi/errors v0.22.0 // indirect
@@ -309,26 +309,26 @@ require (
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.60.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/net/http/httptrace/otelhttptrace v0.59.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.60.0 // indirect
-	go.opentelemetry.io/otel v1.36.0 // indirect
-	go.opentelemetry.io/otel/metric v1.36.0 // indirect
-	go.opentelemetry.io/otel/sdk v1.35.0 // indirect
+	go.opentelemetry.io/otel v1.37.0 // indirect
+	go.opentelemetry.io/otel/metric v1.37.0 // indirect
+	go.opentelemetry.io/otel/sdk v1.36.0 // indirect
 	go.opentelemetry.io/otel/sdk/metric v1.35.0 // indirect
-	go.opentelemetry.io/otel/trace v1.36.0 // indirect
+	go.opentelemetry.io/otel/trace v1.37.0 // indirect
 	go.uber.org/atomic v1.11.0 // indirect
 	go.uber.org/goleak v1.3.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	go.uber.org/zap v1.27.0 // indirect
 	golang.org/x/crypto v0.39.0 // indirect
 	golang.org/x/exp v0.0.0-20250506013437-ce4c2cf36ca6 // indirect
-	golang.org/x/mod v0.25.0 // indirect
+	golang.org/x/mod v0.26.0 // indirect
 	golang.org/x/net v0.41.0 // indirect
 	golang.org/x/oauth2 v0.29.0 // indirect
-	golang.org/x/sync v0.15.0 // indirect
+	golang.org/x/sync v0.16.0 // indirect
 	golang.org/x/sys v0.33.0 // indirect
 	golang.org/x/term v0.32.0 // indirect
-	golang.org/x/text v0.26.0 // indirect
+	golang.org/x/text v0.27.0 // indirect
 	golang.org/x/time v0.11.0 // indirect
-	golang.org/x/tools v0.33.0 // indirect
+	golang.org/x/tools v0.34.0 // indirect
 	golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da // indirect
 	google.golang.org/api v0.227.0 // indirect
 	google.golang.org/genproto v0.0.0-20250324211829-b45e905df463 // indirect
diff --git a/go.sum b/go.sum
index 8fb3c00..5ba41f4 100644
--- a/go.sum
+++ b/go.sum
@@ -661,18 +661,18 @@ github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03
 github.com/BurntSushi/toml v1.4.1-0.20240526193622-a339e1f7089c h1:pxW6RcqyfI9/kWtOwnv/G+AzdKuy2ZrqINhenH4HyNs=
 github.com/BurntSushi/toml v1.4.1-0.20240526193622-a339e1f7089c/go.mod h1:ukJfTF/6rtPPRCnwkur4qwRxa8vTRFBF0uk2lLoLwho=
 github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo=
-github.com/ClickHouse/ch-go v0.66.0 h1:hLslxxAVb2PHpbHr4n0d6aP8CEIpUYGMVT1Yj/Q5Img=
-github.com/ClickHouse/ch-go v0.66.0/go.mod h1:noiHWyLMJAZ5wYuq3R/K0TcRhrNA8h7o1AqHX0klEhM=
-github.com/ClickHouse/clickhouse-go/v2 v2.37.0 h1:Sk7760AFd+mQBN1ccj5/C+WpkWjnvhL8h5m8topl2UQ=
-github.com/ClickHouse/clickhouse-go/v2 v2.37.0/go.mod h1:1KKjGFSWu2R/oa7DKWJLlhTOtyCld7VJDEtXTe+2QKU=
+github.com/ClickHouse/ch-go v0.66.1 h1:LQHFslfVYZsISOY0dnOYOXGkOUvpv376CCm8g7W74A4=
+github.com/ClickHouse/ch-go v0.66.1/go.mod h1:NEYcg3aOFv2EmTJfo4m2WF7sHB/YFbLUuIWv9iq76xY=
+github.com/ClickHouse/clickhouse-go/v2 v2.37.2 h1:wRLNKoynvHQEN4znnVHNLaYnrqVc9sGJmGYg+GGCfto=
+github.com/ClickHouse/clickhouse-go/v2 v2.37.2/go.mod h1:pH2zrBGp5Y438DMwAxXMm1neSXPPjSI7tD4MURVULw8=
 github.com/Code-Hex/go-generics-cache v1.5.1 h1:6vhZGc5M7Y/YD8cIUcY8kcuQLB4cHR7U+0KMqAA0KcU=
 github.com/Code-Hex/go-generics-cache v1.5.1/go.mod h1:qxcC9kRVrct9rHeiYpFWSoW1vxyillCVzX13KZG8dl4=
-github.com/DIMO-Network/clickhouse-infra v0.0.3 h1:B6/4IY9IxLcyydET14IjHUT+A5SDEis7p//DoFzrMk4=
-github.com/DIMO-Network/clickhouse-infra v0.0.3/go.mod h1:NtpQ1btkPzebDvpYYygeqiiBmJ/q5oJb/T/JWzUVRlk=
+github.com/DIMO-Network/clickhouse-infra v0.0.4 h1:Mff+m2hGC5vrI1kdD3oUKt7bAbwM2IadR3w/24Ocu5E=
+github.com/DIMO-Network/clickhouse-infra v0.0.4/go.mod h1:TBW6gZPV/9Z364UGbvVSYShowhKZKwlS+raADwMgiDA=
 github.com/DIMO-Network/cloudevent v0.1.2 h1:6T62P5EKj0IAtN4Z8GQ1tRucmoco29jzyzKAzSrW0cY=
 github.com/DIMO-Network/cloudevent v0.1.2/go.mod h1:HM9vpx4EmJv3Dh8GIXmNhmvHFLGZHpmlv1LXo98oM5w=
-github.com/DIMO-Network/model-garage v0.6.0 h1:VtRjcEljCTffeDQbq5k8LP1Y01FBt5OpgwXRrH5QiPY=
-github.com/DIMO-Network/model-garage v0.6.0/go.mod h1:5xrWBU0Gr6l3szBtfwM76KbsQlT6GTX2XI+4s/ZD9O0=
+github.com/DIMO-Network/model-garage v0.6.5 h1:De+RIyIMVM4b8yvtU6RNiON2vsZHsbLB6yIwu0Lu8Ww=
+github.com/DIMO-Network/model-garage v0.6.5/go.mod h1:89G/TNoTM/MGMof9PX/wdF1KIbthib/Wb1LB8BOJBKY=
 github.com/DataDog/datadog-go v3.2.0+incompatible/go.mod h1:LButxg5PwREeZtORoXG3tL4fMGNddJ+vMq1mwgfaqoQ=
 github.com/GoogleCloudPlatform/grpc-gcp-go/grpcgcp v1.5.2 h1:DBjmt6/otSdULyJdVg2BlG0qGZO5tKL4VzOs0jpvw5Q=
 github.com/GoogleCloudPlatform/grpc-gcp-go/grpcgcp v1.5.2/go.mod h1:dppbR7CwXD4pgtV9t3wD1812RaLDcBjtblcDF5f1vI0=
@@ -718,8 +718,8 @@ github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRF
 github.com/alecthomas/units v0.0.0-20240927000941-0f3dac36c52b h1:mimo19zliBX/vSQ6PWWSL9lK8qwHozUj03+zLoEB8O0=
 github.com/alecthomas/units v0.0.0-20240927000941-0f3dac36c52b/go.mod h1:fvzegU4vN3H1qMT+8wDmzjAcDONcgo2/SZ/TyfdUOFs=
 github.com/andybalholm/brotli v1.0.4/go.mod h1:fO7iG3H7G2nSZ7m0zPUDn85XEX2GTukHGRSepvi9Eig=
-github.com/andybalholm/brotli v1.1.1 h1:PR2pgnyFznKEugtsUo0xLdDop5SKXd5Qf5ysW+7XdTA=
-github.com/andybalholm/brotli v1.1.1/go.mod h1:05ib4cKhjx3OQYUY22hTVd34Bc8upXjOLL2rKwwZBoA=
+github.com/andybalholm/brotli v1.2.0 h1:ukwgCxwYrmACq68yiUqwIWnGY0cTPox/M94sVwToPjQ=
+github.com/andybalholm/brotli v1.2.0/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY=
 github.com/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY=
 github.com/apache/arrow-go/v18 v18.0.0 h1:1dBDaSbH3LtulTyOVYaBCHO3yVRwjV+TZaqn3g6V7ZM=
 github.com/apache/arrow-go/v18 v18.0.0/go.mod h1:t6+cWRSmKgdQ6HsxisQjok+jBpKGhRDiqcf3p0p/F+A=
@@ -955,8 +955,8 @@ github.com/envoyproxy/protoc-gen-validate v0.9.1/go.mod h1:OKNgG7TCp5pF4d6XftA0+
 github.com/envoyproxy/protoc-gen-validate v0.10.1/go.mod h1:DRjgyB0I43LtJapqN6NiRwroiAU2PaFuvk/vjgh61ss=
 github.com/envoyproxy/protoc-gen-validate v1.2.1 h1:DEo3O99U8j4hBFwbJfrz9VtgcDfUKS7KJ7spH3d86P8=
 github.com/envoyproxy/protoc-gen-validate v1.2.1/go.mod h1:d/C80l/jxXLdfEIhX1W2TmLfsJ31lvEjwamM4DxlWXU=
-github.com/ethereum/go-ethereum v1.15.11 h1:JK73WKeu0WC0O1eyX+mdQAVHUV+UR1a9VB/domDngBU=
-github.com/ethereum/go-ethereum v1.15.11/go.mod h1:mf8YiHIb0GR4x4TipcvBUPxJLw1mFdmxzoDi11sDRoI=
+github.com/ethereum/go-ethereum v1.16.1 h1:7684NfKCb1+IChudzdKyZJ12l1Tq4ybPZOITiCDXqCk=
+github.com/ethereum/go-ethereum v1.16.1/go.mod h1:ngYIvmMAYdo4sGW9cGzLvSsPGhDOOzL0jK5S5iXpj0g=
 github.com/facette/natsort v0.0.0-20181210072756-2cd4dd1e2dcb h1:IT4JYU7k4ikYg1SCxNI1/Tieq/NFvh6dzLdgi7eu0tM=
 github.com/facette/natsort v0.0.0-20181210072756-2cd4dd1e2dcb/go.mod h1:bH6Xx7IW64qjjJq8M2u4dxNaBiDfKK+z/3eGDpXEQhc=
 github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
@@ -1000,8 +1000,8 @@ github.com/go-latex/latex v0.0.0-20210823091927-c0d11ff05a81/go.mod h1:SX0U8uGpx
 github.com/go-logfmt/logfmt v0.3.0/go.mod h1:Qt1PoO58o5twSAckw1HlFXLmHsOX5/0LbT9GBnD5lWE=
 github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V4qmtdjCk=
 github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
-github.com/go-logr/logr v1.4.2 h1:6pFjapn8bFcIbiKo3XT4j/BhANplGihG6tvd+8rYgrY=
-github.com/go-logr/logr v1.4.2/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
+github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI=
+github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
 github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
 github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
 github.com/go-ole/go-ole v1.3.0 h1:Dt6ye7+vXGIKZ7Xtk4s6/xVdGDQynvom7xCFEdWr6uE=
@@ -1689,8 +1689,8 @@ github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOf
 github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
 github.com/testcontainers/testcontainers-go v0.37.0 h1:L2Qc0vkTw2EHWQ08djon0D2uw7Z/PtHS/QzZZ5Ra/hg=
 github.com/testcontainers/testcontainers-go v0.37.0/go.mod h1:QPzbxZhQ6Bclip9igjLFj6z0hs01bU8lrl2dHQmgFGM=
-github.com/testcontainers/testcontainers-go/modules/clickhouse v0.33.0 h1:YbB5DBkpgY+GlGPFqTSV1hzWPm3ZHirEyooZrj+ZXK0=
-github.com/testcontainers/testcontainers-go/modules/clickhouse v0.33.0/go.mod h1:qJuMPl9yWIWasmdBILM2uDk1Ny1kdeigcKMJ6A8PZz0=
+github.com/testcontainers/testcontainers-go/modules/clickhouse v0.35.0 h1:A4NESGwof4RK+i/pjL0lAVu0JxNZIvLR35ZFB9DKgYQ=
+github.com/testcontainers/testcontainers-go/modules/clickhouse v0.35.0/go.mod h1:nT0LQ4rqTljX5Ub0Q3GdFrVXRYSrjK6p7RuJPpUE4wg=
 github.com/tidwall/gjson v1.14.2/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk=
 github.com/tidwall/gjson v1.18.0 h1:FIDeeyB800efLX89e5a8Y0BNH+LOngJyGrIWxG2FKQY=
 github.com/tidwall/gjson v1.18.0/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk=
@@ -1840,20 +1840,20 @@ go.opentelemetry.io/contrib/instrumentation/net/http/httptrace/otelhttptrace v0.
 go.opentelemetry.io/contrib/instrumentation/net/http/httptrace/otelhttptrace v0.59.0/go.mod h1:54CaSNqYEXvpzDh8KPjiMVoWm60t5R0dZRt0leEPgAs=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.60.0 h1:sbiXRNDSWJOTobXh5HyQKjq6wUC5tNybqjIqDpAY4CU=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.60.0/go.mod h1:69uWxva0WgAA/4bu2Yy70SLDBwZXuQ6PbBpbsa5iZrQ=
-go.opentelemetry.io/otel v1.36.0 h1:UumtzIklRBY6cI/lllNZlALOF5nNIzJVb16APdvgTXg=
-go.opentelemetry.io/otel v1.36.0/go.mod h1:/TcFMXYjyRNh8khOAO9ybYkqaDBb/70aVwkNML4pP8E=
+go.opentelemetry.io/otel v1.37.0 h1:9zhNfelUvx0KBfu/gb+ZgeAfAgtWrfHJZcAqFC228wQ=
+go.opentelemetry.io/otel v1.37.0/go.mod h1:ehE/umFRLnuLa/vSccNq9oS1ErUlkkK71gMcN34UG8I=
 go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.35.0 h1:1fTNlAIJZGWLP5FVu0fikVry1IsiUnXjf7QFvoNN3Xw=
 go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.35.0/go.mod h1:zjPK58DtkqQFn+YUMbx0M2XV3QgKU0gS9LeGohREyK4=
 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.34.0 h1:BEj3SPM81McUZHYjRS5pEgNgnmzGJ5tRpU5krWnV8Bs=
 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.34.0/go.mod h1:9cKLGBDzI/F3NoHLQGm4ZrYdIHsvGt6ej6hUowxY0J4=
-go.opentelemetry.io/otel/metric v1.36.0 h1:MoWPKVhQvJ+eeXWHFBOPoBOi20jh6Iq2CcCREuTYufE=
-go.opentelemetry.io/otel/metric v1.36.0/go.mod h1:zC7Ks+yeyJt4xig9DEw9kuUFe5C3zLbVjV2PzT6qzbs=
-go.opentelemetry.io/otel/sdk v1.35.0 h1:iPctf8iprVySXSKJffSS79eOjl9pvxV9ZqOWT0QejKY=
-go.opentelemetry.io/otel/sdk v1.35.0/go.mod h1:+ga1bZliga3DxJ3CQGg3updiaAJoNECOgJREo9KHGQg=
+go.opentelemetry.io/otel/metric v1.37.0 h1:mvwbQS5m0tbmqML4NqK+e3aDiO02vsf/WgbsdpcPoZE=
+go.opentelemetry.io/otel/metric v1.37.0/go.mod h1:04wGrZurHYKOc+RKeye86GwKiTb9FKm1WHtO+4EVr2E=
+go.opentelemetry.io/otel/sdk v1.36.0 h1:b6SYIuLRs88ztox4EyrvRti80uXIFy+Sqzoh9kFULbs=
+go.opentelemetry.io/otel/sdk v1.36.0/go.mod h1:+lC+mTgD+MUWfjJubi2vvXWcVxyr9rmlshZni72pXeY=
 go.opentelemetry.io/otel/sdk/metric v1.35.0 h1:1RriWBmCKgkeHEhM7a2uMjMUfP7MsOF5JpUCaEqEI9o=
 go.opentelemetry.io/otel/sdk/metric v1.35.0/go.mod h1:is6XYCUMpcKi+ZsOvfluY5YstFnhW0BidkR+gL+qN+w=
-go.opentelemetry.io/otel/trace v1.36.0 h1:ahxWNuqZjpdiFAyrIoQ4GIiAIhxAunQR6MUoKrsNd4w=
-go.opentelemetry.io/otel/trace v1.36.0/go.mod h1:gQ+OnDZzrybY4k4seLzPAWNwVBBVlF2szhehOBB/tGA=
+go.opentelemetry.io/otel/trace v1.37.0 h1:HLdcFNbRQBE2imdSEgm/kwqmQj1Or1l/7bW6mxVK7z4=
+go.opentelemetry.io/otel/trace v1.37.0/go.mod h1:TlgrlQ+PtQO5XFerSPUYG0JSgGyryXewPGyayAWSBS0=
 go.opentelemetry.io/proto/otlp v0.7.0/go.mod h1:PqfVotwruBrMGOCsRd/89rSnXhoiJIqeYNgFYFoEGnI=
 go.opentelemetry.io/proto/otlp v0.15.0/go.mod h1:H7XAot3MsfNsj7EXtrA2q5xSNQ10UqI405h3+duxN4U=
 go.opentelemetry.io/proto/otlp v0.19.0/go.mod h1:H7XAot3MsfNsj7EXtrA2q5xSNQ10UqI405h3+duxN4U=
@@ -1942,8 +1942,8 @@ golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91
 golang.org/x/mod v0.7.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
 golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
 golang.org/x/mod v0.9.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
-golang.org/x/mod v0.25.0 h1:n7a+ZbQKQA/Ysbyb0/6IbB1H/X41mKgbhfv7AfG/44w=
-golang.org/x/mod v0.25.0/go.mod h1:IXM97Txy2VM4PJ3gI61r1YEk/gAj6zAHN3AdZt6S9Ww=
+golang.org/x/mod v0.26.0 h1:EGMPT//Ezu+ylkCijjPc+f4Aih7sZvaAr+O3EHBxvZg=
+golang.org/x/mod v0.26.0/go.mod h1:/j6NAhSk8iQ723BGAUyoAcn7SlD7s15Dp9Nd/SfeaFQ=
 golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20181114220301-adae6a3d119a/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
@@ -2056,8 +2056,8 @@ golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20220819030929-7fc1605a5dde/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20220929204114-8fcdb60fdcc0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.15.0 h1:KWH3jNZsfyT6xfAfKiz6MRNmd46ByHDYaZ7KSkCtdW8=
-golang.org/x/sync v0.15.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=
+golang.org/x/sync v0.16.0 h1:ycBJEhp9p4vXvUZNszeOq0kGTPghopOL8q0fq3vstxw=
+golang.org/x/sync v0.16.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=
 golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -2175,8 +2175,8 @@ golang.org/x/text v0.6.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/text v0.8.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
 golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
-golang.org/x/text v0.26.0 h1:P42AVeLghgTYr4+xUnTRKDMqpar+PtX7KWuNQL21L8M=
-golang.org/x/text v0.26.0/go.mod h1:QK15LZJUUQVJxhz7wXgxSy/CJaTFjd0G+YLonydOVQA=
+golang.org/x/text v0.27.0 h1:4fGWRpyh641NLlecmyl4LOe6yDdfaYNrGb2zdfo4JV4=
+golang.org/x/text v0.27.0/go.mod h1:1D28KMCvyooCX9hBiosv5Tz/+YLxj0j7XhWjpSUF7CU=
 golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
@@ -2249,8 +2249,8 @@ golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc
 golang.org/x/tools v0.3.0/go.mod h1:/rWhSS2+zyEVwoJf8YAX6L2f0ntZ7Kn/mGgAWcipA5k=
 golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU=
 golang.org/x/tools v0.7.0/go.mod h1:4pg6aUX35JBAogB10C9AtvVL+qowtN4pT3CGSQex14s=
-golang.org/x/tools v0.33.0 h1:4qz2S3zmRxbGIhDIAgjxvFutSvH5EfnsYrRBj0UI0bc=
-golang.org/x/tools v0.33.0/go.mod h1:CIJMaWEY88juyUfo7UbgPqbC8rU2OqfAV1h2Qp0oMYI=
+golang.org/x/tools v0.34.0 h1:qIpSLOxeCYGg9TrcJokLBG4KFA6d795g0xkBkiESGlo=
+golang.org/x/tools v0.34.0/go.mod h1:pAP9OwEaY1CAW3HOmg3hLZC5Z0CCmzjAF2UQMSqNARg=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
diff --git a/internal/processors/dbmigration/processors.go b/internal/processors/dbmigration/processors.go
index 892f059..6fc0e76 100644
--- a/internal/processors/dbmigration/processors.go
+++ b/internal/processors/dbmigration/processors.go
@@ -13,8 +13,8 @@ import (
 )
 
 const (
-	indexMigrationProcName  = "dimo_file_index_migration"
-	signalMigrationProcName = "dimo_signal_migration"
+	indexMigrationProcName  = "dimo_file_index_db_migration"
+	signalMigrationProcName = "dimo_db_migration"
 )
 
 var configSpec = service.NewConfigSpec().
diff --git a/internal/processors/eventstoslice/eventstoslice.go b/internal/processors/eventstoslice/eventstoslice.go
new file mode 100644
index 0000000..9bf1734
--- /dev/null
+++ b/internal/processors/eventstoslice/eventstoslice.go
@@ -0,0 +1,60 @@
+package eventstoslice
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+
+	"github.com/DIMO-Network/model-garage/pkg/vss"
+	"github.com/redpanda-data/benthos/v4/public/service"
+)
+
+type eventSliceProcessor struct {
+	logger *service.Logger
+}
+
+func init() {
+	// Config spec is empty for now as we don't have any dynamic fields.
+	configSpec := service.NewConfigSpec().Description("Event Json Object to slice.")
+	constructor := func(_ *service.ParsedConfig, mgr *service.Resources) (service.Processor, error) {
+		return newEventSliceProcessor(mgr.Logger()), nil
+	}
+	err := service.RegisterProcessor("dimo_event_to_slice", configSpec, constructor)
+	if err != nil {
+		panic(err)
+	}
+}
+
+func newEventSliceProcessor(lgr *service.Logger) *eventSliceProcessor {
+	// The logger will already be labelled with the
+	// identifier of this component within a config.
+	return &eventSliceProcessor{
+		logger: lgr,
+	}
+}
+
+func (s *eventSliceProcessor) Process(_ context.Context, msg *service.Message) (service.MessageBatch, error) {
+	// Extract the message payload as a byte slice.
+	payload, err := msg.AsBytes()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get event message payload: %w", err)
+	}
+
+	var event []vss.Event
+	err = json.Unmarshal(payload, &event)
+	if err != nil {
+		return nil, fmt.Errorf("failed to unmarshal event: %w", err)
+	}
+	msgs := make([]*service.Message, 0, len(event))
+	for _, e := range event {
+		sig := vss.EventToSlice(e)
+		msgCpy := msg.Copy()
+		msgCpy.SetStructured(sig)
+		msgs = append(msgs, msgCpy)
+	}
+	return msgs, nil
+}
+
+func (p *eventSliceProcessor) Close(ctx context.Context) error {
+	return nil
+}
diff --git a/internal/processors/signalstoslice/signaltoslice.go b/internal/processors/signalstoslice/signalstoslice.go
similarity index 83%
rename from internal/processors/signalstoslice/signaltoslice.go
rename to internal/processors/signalstoslice/signalstoslice.go
index d9402df..c2d6e51 100644
--- a/internal/processors/signalstoslice/signaltoslice.go
+++ b/internal/processors/signalstoslice/signalstoslice.go
@@ -1,9 +1,10 @@
-package checksignature
+package signalstoslice
 
 import (
 	"context"
 	"encoding/json"
 	"fmt"
+
 	"github.com/DIMO-Network/model-garage/pkg/vss"
 	"github.com/redpanda-data/benthos/v4/public/service"
 )
@@ -18,7 +19,7 @@ func init() {
 	constructor := func(_ *service.ParsedConfig, mgr *service.Resources) (service.Processor, error) {
 		return newSliceProcessor(mgr.Logger()), nil
 	}
-	err := service.RegisterProcessor("signal_to_slice", configSpec, constructor)
+	err := service.RegisterProcessor("dimo_signal_to_slice", configSpec, constructor)
 	if err != nil {
 		panic(err)
 	}
@@ -36,14 +37,13 @@ func (s *sliceProcessor) Process(_ context.Context, msg *service.Message) (servi
 	// Extract the message payload as a byte slice.
 	payload, err := msg.AsBytes()
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("failed to get signal message payload: %w", err)
 	}
 
 	var signal vss.Signal
 	err = json.Unmarshal(payload, &signal)
 	if err != nil {
-		fmt.Println(err)
-		return nil, err
+		return nil, fmt.Errorf("failed to unmarshal signal: %w", err)
 	}
 
 	sig := vss.SignalToSlice(signal)
diff --git a/main.go b/main.go
index 22f05ed..24dfd20 100644
--- a/main.go
+++ b/main.go
@@ -22,6 +22,7 @@ import (
 
 	// Add our custom plugin packages here.
 	_ "github.com/DIMO-Network/dps/internal/processors/dbmigration"
+	_ "github.com/DIMO-Network/dps/internal/processors/eventstoslice"
 	_ "github.com/DIMO-Network/dps/internal/processors/signalstoslice"
 	_ "github.com/DIMO-Network/dps/internal/processors/splitvalues"
 )