Remove IPFS anchoring

.env.example

@@ -5,7 +5,6 @@ KAFKA_BROKERS=localhost:9092
 KAFKA_CLIENT_ID=sensors-connectivity
 
 REDIS_URL=redis://localhost:6379
-IPFS_API_URL=http://localhost:5001
 SUBSTRATE_WS_URL=ws://localhost:9944
 
 ENDPOINT_PORT=3000
@@ -29,30 +28,18 @@ HEARTBEAT_TRACKER_SOURCE=heartbeat-tracker
 HEARTBEAT_TRACKER_HEALTH_PORT=3030
 HEARTBEAT_TRACKER_ONLINE_WINDOW_MS=30000
 HEARTBEAT_TRACKER_REDIS_PREFIX=heartbeat-tracker:v1
-# Batcher (batches authorized telemetry into telemetry.batched.v1)
+# Batcher (batches, XZ-compresses, and size-fits authorized telemetry into telemetry.batched.v1)
 BATCHER_GROUP_ID=batcher-v1
 BATCHER_SOURCE=batcher
 BATCHER_HEALTH_PORT=3041
 BATCHER_BATCH_SIZE=10
 BATCHER_BATCH_TIMEOUT_MS=30000
-# IPFS publisher (publishes telemetry.batched.v1 batches to IPFS)
-IPFS_PUBLISHER_GROUP_ID=ipfs-publisher-v1
-IPFS_PUBLISHER_SOURCE=ipfs-publisher
-IPFS_PUBLISHER_HEALTH_PORT=3040
-IPFS_PUBLISHER_ENABLE_COMPRESSION=true
-# Ordered, comma-separated provider list; first entry is primary (e.g. pinata,kubo)
-IPFS_PROVIDERS=kubo
-# PINATA_API_URL=https://api.pinata.cloud
-# PINATA_JWT=
-# Durability policy: any (default) | all | quorum
-IPFS_DURABILITY_POLICY=any
-IPFS_DURABILITY_MIN_SUCCESS_COUNT=1
-IPFS_PROVIDER_RETRY_BASE_DELAY_MS=5000
-IPFS_PROVIDER_RETRY_MAX_DELAY_MS=60000
-IPFS_PROVIDER_REPLICATION_MAX_PENDING=1000
-# Blockchain anchor (anchors IPFS CIDs to Robonomics CPS pallet)
+# Shared max compressed payload size for a single anchored batch (bytes)
+ANCHOR_MAX_PAYLOAD_BYTES=8192
+# Blockchain anchor (anchors compressed telemetry batches to Robonomics CPS pallet)
 BLOCKCHAIN_ANCHOR_GROUP_ID=blockchain-anchor-v1
 BLOCKCHAIN_ANCHOR_NODE_ID=0
 BLOCKCHAIN_ANCHOR_SURI=//Alice
 BLOCKCHAIN_ANCHOR_HEALTH_PORT=3050
+BLOCKCHAIN_ANCHOR_SOURCE=blockchain-anchor
 BLOCKCHAIN_TARGET=robonomics-dev

.github/copilot-instructions.md

@@ -39,10 +39,9 @@ This is an **event-driven telemetry pipeline** with Kafka as the central durable
 4. Multiple **downstream consumers** process authorized events independently:
    - **pubsub-broadcaster**: Publishes to libp2p/GossipSub for real-time web UI
    - **heartbeat-tracker**: Tracks sensor liveness/uptime metrics (observability-only, no DLQ)
-   - **batcher**: Batches authorized telemetry and emits `telemetry.batched.v1`
-5. **batcher** → **ipfs-publisher** → **blockchain-anchor** publication chain:
-   - **ipfs-publisher**: Consumes `telemetry.batched.v1`, publishes batches to IPFS, emits CIDs
-   - **blockchain-anchor**: Anchors IPFS CIDs to Robonomics blockchain
+   - **batcher**: Batches authorized telemetry, XZ-compresses and splits it to fit `ANCHOR_MAX_PAYLOAD_BYTES`, and emits `telemetry.batched.v1`
+5. **batcher** → **blockchain-anchor** publication chain:
+   - **blockchain-anchor**: Consumes `telemetry.batched.v1`, submits the compressed payload via `cps.setPayload` to anchor it on the Robonomics blockchain, deduplicating by comparing against the current on-chain payload
 
 ### Key Architectural Constraints
 - **No direct coupling** between processing modules (all flow through Kafka)
@@ -60,8 +59,7 @@ Robonomics Blockchain → registry-sync → Redis → endpoint (lookup during va
 ### Core Kafka Topics
 - `telemetry.authorized.v1` - Successfully validated telemetry
 - `telemetry.rejected.v1` - Failed validation (signature/timestamp/auth)
-- `telemetry.batched.v1` - Batched authorized telemetry ready for publication
-- `ipfs.published.v1` - IPFS publish results (includes CID)
+- `telemetry.batched.v1` - XZ-compressed, chain-ready batched telemetry (submitted directly via `cps.setPayload`)
 - `telemetry.blockchain.result.v1` - Blockchain anchoring results
 - `telemetry.retry.v1` - Transient failures for retry
 - `telemetry.dlq.v1` - Exhausted retries (dead letters)
@@ -73,9 +71,8 @@ Robonomics Blockchain → registry-sync → Redis → endpoint (lookup during va
 - `services/whitelist` - Whitelist-based sensor auth provider
 - `services/pubsub-broadcaster` - Kafka→libp2p GossipSub bridge
 - `services/heartbeat-tracker` - Observability metrics (online sensors, uptime)
-- `services/batcher` - Kafka batcher (authorized → `telemetry.batched.v1`)
-- `services/ipfs-publisher` - Kafka→IPFS publisher (Kubo RPC)
-- `services/blockchain-anchor` - IPFS CID→blockchain anchoring
+- `services/batcher` - Kafka batcher (authorized → XZ-compressed, size-fitted `telemetry.batched.v1`)
+- `services/blockchain-anchor` - Anchors compressed batch payloads to the Robonomics CPS pallet via `cps.setPayload`
 - `tools/fake-sensor-cli` - Generate test telemetry with Ed25519 signatures
 
 ## Key Conventions and Patterns
@@ -112,7 +109,7 @@ All Kafka events follow a strict envelope schema (see `packages/contracts/src/en
 All Kafka consumers follow the same processing rule (see `packages/contracts/src/consumer-runtime.ts`):
 1. Consume event
 2. Check deduplication (if applicable)
-3. Perform external action (e.g., publish to PubSub/IPFS)
+3. Perform external action (e.g., publish to PubSub, submit blockchain extrinsic)
 4. Wait for confirmation
 5. Emit result event
 6. Commit offset
@@ -127,7 +124,7 @@ Use `runConsumerProcessingRule()` helper with:
 - Services load config from environment variables with defaults
 - Common pattern: `load<Service>Config()` function validates and returns typed config
 - Use zod schemas for validation where possible
-- Required infrastructure endpoints: Kafka, Redis, IPFS Kubo RPC, Robonomics node
+- Required infrastructure endpoints: Kafka, Redis, Robonomics node
 
 ### Naming Conventions
 - Services: `@scp/<name>` (e.g., `@scp/endpoint`)
@@ -152,7 +149,7 @@ Use `runConsumerProcessingRule()` helper with:
 1. Prerequisites: Node.js 20 (`.nvmrc`), pnpm 9+, Docker Compose
 2. `pnpm install`
 3. `cp .env.example .env` and configure as needed
-4. `docker compose up -d` (starts Kafka, Redis, IPFS, Robonomics node)
+4. `docker compose up -d` (starts Kafka, Redis, Robonomics node)
 5. `pnpm dev` (starts all services) or `pnpm --filter @scp/<name> dev` (single service)
 
 ## Testing Philosophy

.github/workflows/integration-test.yml

@@ -116,7 +116,6 @@ jobs:
           (pnpm --filter @scp/registry-sync dev > logs/registry-sync.log 2>&1 & echo $! > logs/registry-sync.pid)
           (pnpm --filter @scp/endpoint dev > logs/endpoint.log 2>&1 & echo $! > logs/endpoint.pid)
           (PUBSUB_RESERVED_PEERS="${PUBSUB_RESERVED_PEERS}" pnpm --filter @scp/pubsub-broadcaster dev > logs/pubsub-broadcaster.log 2>&1 & echo $! > logs/pubsub-broadcaster.pid)
-          (pnpm --filter @scp/ipfs-publisher dev > logs/ipfs-publisher.log 2>&1 & echo $! > logs/ipfs-publisher.pid)
           (pnpm --filter @scp/blockchain-anchor dev > logs/blockchain-anchor.log 2>&1 & echo $! > logs/blockchain-anchor.pid)
 
       - name: Wait for service health endpoints
@@ -216,7 +215,7 @@ jobs:
           fi
 
           if grep -Eqi '(^|[^a-z])(error|exception|unhandledrejection|econnrefused|failed to start|failed to publish)([^a-z]|$)' \
-            logs/endpoint.log logs/registry-sync.log logs/pubsub-broadcaster.log logs/ipfs-publisher.log logs/blockchain-anchor.log; then
+            logs/endpoint.log logs/registry-sync.log logs/pubsub-broadcaster.log logs/blockchain-anchor.log; then
             echo 'Detected error markers in service logs' >&2
             exit 1
           fi

README.md

@@ -23,7 +23,7 @@ make all
 This will:
 1. Install dependencies with pnpm
 2. Build all packages and services
-3. Start infrastructure (Kafka, Redis, IPFS, Robonomics)
+3. Start infrastructure (Kafka, Redis, Robonomics)
 4. Initialize Kafka topics
 5. Start all services
 
@@ -36,9 +36,8 @@ services/registry-sync     # Robonomics blockchain→Redis projection sync servi
 services/whitelist         # Whitelist-based sensor authentication provider
 services/pubsub-broadcaster # Kafka→libp2p GossipSub bridge for real-time web UI
 services/heartbeat-tracker  # Observability: sensor liveness & uptime metrics
-services/batcher           # Kafka: batches authorized telemetry into telemetry.batched.v1
-services/ipfs-publisher    # Kafka→IPFS publisher (publishes batches, produces CIDs)
-services/blockchain-anchor # IPFS CID→Robonomics CPS pallet anchoring
+services/batcher           # Kafka: batches, compresses (XZ), and splits telemetry into telemetry.batched.v1
+services/blockchain-anchor # Anchors chain-ready compressed batches to the Robonomics CPS pallet
 tools/fake-sensor-cli      # Generate test telemetry with Ed25519 signatures
 ```
 
@@ -95,6 +94,5 @@ Use `REJECTION_CODES` from `@scp/core` to reference these codes in your code:
 - **whitelist**: Alternative authentication provider - maintains static sensor whitelist in Redis, bypassing blockchain dependency for simpler deployments.
 - **pubsub-broadcaster**: Consumes `telemetry.authorized.v1`, publishes to libp2p/GossipSub for real-time web UI, emits `telemetry.pubsub.result.v1`, routes exhausted failures to DLQ.
 - **heartbeat-tracker**: Observability-only consumer of `telemetry.authorized.v1`, tracks sensor liveness (`firstSeen`, `lastSeen`, `onlineSince`) in Redis, exposes `sensors_online` count and per-sensor/aggregate uptime metrics over configurable online window (default 30s). Does not emit result events or participate in retry/DLQ.
-- **batcher**: Consumes `telemetry.authorized.v1`, groups events into deterministic batches (by size, lag, and a bounded flush timer), and emits `telemetry.batched.v1`. Serializes flushes (single active flush per instance) and flushes pending batches on graceful shutdown.
-- **ipfs-publisher**: Consumes `telemetry.batched.v1`, publishes/pins each batch to IPFS (optionally XZ-compressed), deduplicates by `batch_id`, emits `ipfs.published.v1`.
-- **blockchain-anchor**: Consumes `ipfs.published.v1` and anchors CIDs to the Robonomics blockchain.
+- **batcher**: Consumes `telemetry.authorized.v1`, groups events into deterministic batches (by size, lag, and a bounded flush timer), XZ-compresses and splits each batch to fit `ANCHOR_MAX_PAYLOAD_BYTES`, and emits `telemetry.batched.v1`. Serializes flushes (single active flush per instance) and flushes pending batches on graceful shutdown.
+- **blockchain-anchor**: Consumes `telemetry.batched.v1` and anchors the compressed payload bytes to the Robonomics blockchain via `cps.setPayload`, deduplicating by comparing against the current on-chain payload.

docker-compose.yml

@@ -38,6 +38,8 @@ services:
       timeout: 5s
       retries: 5
 
+  # Used only as a libp2p GossipSub reserved peer for pubsub-broadcaster
+  # integration testing (no longer used for telemetry batch storage).
   ipfs:
     image: ipfs/kubo:v0.42.0
     ports:

docker/kafka-init-topics.sh

@@ -6,7 +6,6 @@ TOPICS=(
   "telemetry.authorized.v1"
   "telemetry.rejected.v1"
   "telemetry.batched.v1"
-  "ipfs.published.v1"
   "telemetry.dlq.v1"
 )
 

docs/architecture/project-architecture.md

@@ -6,21 +6,21 @@ The system accepts Ed25519-signed environmental sensor telemetry (Altruist-serie
 
 ## High-level architecture
 
-`Sensor -> Endpoint -> Message Bus (Kafka) -> Services (IPFS, PubSub, Blockchain)`
+`Sensor -> Endpoint -> Message Bus (Kafka) -> Services (PubSub, Batcher, Blockchain Anchor)`
 
 > Authorization source: `Robonomics Blockchain -> Registry Sync -> Redis -> Endpoint` or `Whitelist -> Redis -> Endpoint`
 
 ### PubSub Broadcast
 
-`Trusted Messages (Kafka) -> PubSub Broadcaster -> IPFS PubSub -> Web UI (sensors.social)`
+`Trusted Messages (Kafka) -> PubSub Broadcaster -> libp2p GossipSub -> Web UI (sensors.social)`
 
 ### Heartbeat observability
 
 `Trusted Messages (Kafka) -> Heartbeat Tracker -> /metrics (liveness + uptime)`
 
 ### Blockchain anchoring
 
-`Trusted Messages (Kafka) -> Batcher -> Message Bus (Kafka) -> IPFS Publisher -> Message Bus (Kafka) -> Robonomics Blockchain`
+`Trusted Messages (Kafka) -> Batcher -> Message Bus (Kafka) -> Blockchain Anchor -> Robonomics Blockchain`
 
 ## Module responsibilities
 
@@ -68,30 +68,25 @@ The system accepts Ed25519-signed environmental sensor telemetry (Altruist-serie
 ### Batcher
 - Consumes authorized events from `telemetry.authorized.v1`.
 - Groups events into deterministic batches by size, consumer lag, and a bounded flush timer.
-- Serializes each batch as a `crypto.v1.SignedEnvelopeBatch` and emits `telemetry.batched.v1`.
+- Serializes each batch as a `crypto.v1.SignedEnvelopeBatch`, XZ-compresses it, and recursively splits it into smaller sub-batches if the compressed result exceeds `ANCHOR_MAX_PAYLOAD_BYTES`.
+- Emits one `telemetry.batched.v1` message per fitted sub-batch, each carrying the compressed payload, its size/hash fields, and its own `batch_id`.
+- Routes single events that cannot fit even alone to `telemetry.dlq.v1` without blocking the rest of the batch.
 - Serializes flushes (single active flush per instance) so a timer-triggered flush cannot publish the same batch as a size/lag-triggered flush.
 - Flushes any pending batch during graceful shutdown before closing resources.
-- Commits `telemetry.authorized.v1` offsets only after the batch is durably produced.
+- Commits `telemetry.authorized.v1` offsets only after all sub-batches are durably produced.
 
-### IPFS Publisher
+### Blockchain Anchor
 - Consumes batched events from `telemetry.batched.v1`.
-- Publishes/pins each batch artifact to IPFS (optionally XZ-compressed) and captures the CID.
-- Deduplicates by `batch_id` to avoid duplicate publication of redelivered batches.
-- Emits `ipfs.published.v1` with `cid` and `event_count`.
-- Commits `telemetry.batched.v1` offsets only after publish success and result emission.
-
-### Robonomics Blockchain
-- Consumes IPFS-published events (`ipfs.published.v1`) from Kafka.
-- Publishes the CID into the substrate-based Robonomics blockchain to make the CID immutable.
-- Deduplicates by CID before submission.
-- Emits anchoring result events (`telemetry.blockchain.result.v1`).
-- Commits offset only after blockchain submission confirmation.
+- Defensively rejects (permanent error, offset committed) any payload exceeding `ANCHOR_MAX_PAYLOAD_BYTES`.
+- Submits the compressed payload bytes directly into the substrate-based Robonomics blockchain via `cps.setPayload` to make the batch immutable.
+- Deduplicates by comparing the current on-chain payload bytes against the incoming payload before submission.
+- Commits offset only after blockchain submission confirmation (or a confirmed idempotent skip/rejection).
+- Emitting an anchoring result event (`telemetry.blockchain.result.v1`) is deferred to a future phase.
 
 ## Core Kafka topics
 - `telemetry.authorized.v1`
 - `telemetry.rejected.v1`
 - `telemetry.batched.v1`
-- `ipfs.published.v1`
 - `telemetry.dlq.v1`
 
 ## Error handling baseline
@@ -104,8 +99,7 @@ The system accepts Ed25519-signed environmental sensor telemetry (Altruist-serie
 - Allowed flow: `Endpoint -> Kafka -> Consumers`.
 - Disallowed direct couplings:
   - Endpoint -> PubSub
-  - Endpoint -> IPFS
-  - PubSub -> IPFS
-  - Batcher -> IPFS (must flow through Kafka)
-  - IPFS -> Blockchain (must flow through Kafka)
+  - Endpoint -> Blockchain Anchor
+  - PubSub -> Batcher
+  - Batcher -> Blockchain Anchor (must flow through Kafka)
 - Authentication is pluggable via `SensorAuth` interface but must use Redis for low-latency lookups.