diff --git a/.github/workflows/test.yaml b/.github/workflows/test.yaml
index c6c1cd3..edf28b4 100644
--- a/.github/workflows/test.yaml
+++ b/.github/workflows/test.yaml
@@ -15,11 +15,35 @@ jobs:
      - name: Check out code
        uses: actions/checkout@v4
 
-     - name: unit tests
+     - name: Set up Go
+       uses: actions/setup-go@v5
+       with:
+         go-version: '1.23'
+
+     - name: Cache Go modules
+       uses: actions/cache@v4
+       with:
+         path: |
+           ~/.cache/go-build
+           ~/go/pkg/mod
+         key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }}
+         restore-keys: |
+           ${{ runner.os }}-go-
+
+     - name: Install dependencies
+       run: go mod download
+
+     - name: Run unit tests
        run: make test
 
   integration:
     runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        storage: [memory, postgres, kafka]
+      fail-fast: false  # Continue testing other storage types even if one fails
+    
+    name: integration-${{ matrix.storage }}
     
     steps:
     - uses: actions/checkout@v4
@@ -47,11 +71,8 @@ jobs:
     - name: Install dependencies
       run: go mod download
 
-    - name: Run unit tests
-      run: make test
-
     - name: Build bridge
       run: make build
 
-    - name: Run integration tests
-      run: make integration-test
+    - name: Run integration tests with ${{ matrix.storage }} storage
+      run: make integration-test-${{ matrix.storage }}
diff --git a/Makefile b/Makefile
index 88aef3c..61ad24c 100644
--- a/Makefile
+++ b/Makefile
@@ -1,6 +1,6 @@
 GOFMT_FILES?=$$(find . -name '*.go' | grep -v vendor | grep -v yacc | grep -v .git)
 
-.PHONY: all imports fmt test test-bridge-sdk clean-bridge-sdk test-all integration-test run stop clean logs status
+.PHONY: all imports fmt test test-bridge-sdk clean-bridge-sdk test-all integration-test integration-test-memory integration-test-postgres integration-test-kafka run stop clean logs status
 
 all: imports fmt test
 
@@ -30,7 +30,19 @@ test-all: test test-bridge-sdk
 	@echo "All tests completed successfully!"
 
 integration-test:
-	@./scripts/integration-test.sh
+	@./scripts/integration-test.sh memory
+
+integration-test-memory:
+	@echo "🧪 Running integration tests with memory storage..."
+	@./scripts/integration-test.sh memory
+
+integration-test-postgres:
+	@echo "🧪 Running integration tests with PostgreSQL storage..."
+	@./scripts/integration-test.sh postgres
+
+integration-test-kafka:
+	@echo "🧪 Running integration tests with Kafka storage..."
+	@./scripts/integration-test.sh kafka
 
 run:
 	@echo "Starting bridge environment with nginx load balancer and 3 bridge instances..."
diff --git a/README.md b/README.md
index ca8495a..33c3bb3 100644
--- a/README.md
+++ b/README.md
@@ -11,6 +11,30 @@ make build
 ./callmebridge
 ```
 
+### PostgreSQL Storage
+```bash
+# Start PostgreSQL and set environment
+export STORAGE_TYPE="postgres"
+export POSTGRES_URI="postgres://user:pass@localhost/bridge"
+./callmebridge
+```
+
+### Kafka Storage
+```bash
+# Start Kafka (using Docker Compose)
+docker-compose -f docker-compose.kafka.yml up -d kafka
+
+# Or use the provided script
+./scripts/run-with-kafka.sh
+```
+
+### Full Multi-Instance Setup with Kafka
+```bash
+# Start the entire stack with load balancer
+docker-compose -f docker-compose.kafka.yml up
+# Access via http://localhost:8080 (nginx load balancer)
+```
+
 ## 📋Requirements
 
 - Go 1.23+
@@ -23,9 +47,56 @@ Configure using environment variables:
 
 ```bash
 PORT=8081                    # Server port
-POSTGRES_URI="postgres://user:pass@host/dbname"  # Database connection
+
+# Storage Configuration
+STORAGE_TYPE="memory"        # Storage type: memory, postgres, kafka
+POSTGRES_URI="postgres://user:pass@host/dbname"  # PostgreSQL connection (for postgres storage)
+
+# Kafka Configuration (for kafka storage)
+KAFKA_BROKERS="localhost:9092,localhost:9093"  # Comma-separated Kafka broker addresses
+KAFKA_TOPIC="bridge-messages"                  # Kafka topic name (default: bridge-messages)
+KAFKA_CONSUMER_GROUP="bridge-consumer"         # Kafka consumer group (default: bridge-consumer)
+
+# Other Configuration
+WEBHOOK_URL=""               # Webhook URL for message forwarding
+COPY_TO_URL=""              # URL to copy messages to
+CORS_ENABLE=false           # Enable CORS
+HEARTBEAT_INTERVAL=10       # Heartbeat interval in seconds
+RPS_LIMIT=1000              # Rate limit (requests per second)
+CONNECTIONS_LIMIT=200       # Maximum concurrent connections
+SELF_SIGNED_TLS=false       # Use self-signed TLS certificate
 ```
 
+### Storage Types
+
+**Memory Storage** (default)
+- Fast, in-memory storage
+- Data is lost on restart
+- Best for development and testing
+
+**PostgreSQL Storage**
+- Persistent storage with automatic migrations
+- Supports high availability setups
+- Automatic cleanup of expired messages
+
+### Kafka Storage
+
+**Kafka Storage** uses modern **KRaft mode** (Kafka Raft metadata mode) instead of Zookeeper:
+
+✅ **KRaft Benefits:**
+- **Simpler architecture**: No separate Zookeeper cluster needed
+- **Faster startup**: Reduced coordination overhead
+- **Better performance**: Lower latency for metadata operations
+- **Easier scaling**: Single service to manage and scale
+- **Production-ready**: Available since Kafka 3.3.0+
+
+**Features:**
+- Distributed, scalable message storage
+- Hybrid approach: messages stored in Kafka with in-memory caching
+- Automatic message expiration and cleanup
+- Supports multiple bridge instances
+- Built-in health checks and monitoring
+
 ## 🛠️API Endpoints
 
 ### Bridge Endpoints
diff --git a/cmd/bridge/main.go b/cmd/bridge/main.go
index 4425806..e8c929c 100644
--- a/cmd/bridge/main.go
+++ b/cmd/bridge/main.go
@@ -81,15 +81,28 @@ func main() {
 	log.Info("Bridge is running")
 	config.LoadConfig()
 
-	dbConn, err := storage.NewStorage(config.Config.DbURI)
+	storageConfig := storage.StorageConfig{
+		Type:               config.Config.StorageType,
+		PostgresURI:        config.Config.DbURI,
+		KafkaBrokers:       config.Config.KafkaBrokers,
+		KafkaTopic:         config.Config.KafkaTopic,
+		KafkaConsumerGroup: config.Config.KafkaConsumerGroup,
+	}
+
+	dbConn, err := storage.NewStorage(storageConfig)
 
 	if err != nil {
 		log.Fatalf("failed to create storage: %v", err)
 	}
-	if _, ok := dbConn.(*storage.MemStorage); ok {
-		log.Info("Using in-memory storage")
-	} else {
+
+	// Log which storage type is being used
+	switch storageConfig.Type {
+	case "kafka":
+		log.Info("Using Kafka storage")
+	case "postgres", "pg":
 		log.Info("Using PostgreSQL storage")
+	default:
+		log.Info("Using in-memory storage")
 	}
 
 	healthMetric.Set(1)
diff --git a/docker-compose.kafka.yml b/docker-compose.kafka.yml
new file mode 100644
index 0000000..d43647e
--- /dev/null
+++ b/docker-compose.kafka.yml
@@ -0,0 +1,145 @@
+version: '3.8'
+
+services:
+  kafka:
+    image: confluentinc/cp-kafka:7.4.0
+    hostname: kafka
+    container_name: kafka
+    ports:
+      - "9092:9092"
+      - "9093:9093"
+      - "9997:9997"
+    environment:
+      # KRaft mode configuration (no Zookeeper needed!)
+      KAFKA_NODE_ID: 1
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT'
+      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092'
+      KAFKA_LISTENERS: 'PLAINTEXT://kafka:29092,CONTROLLER://kafka:29093,PLAINTEXT_HOST://0.0.0.0:9092'
+      KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT'
+      KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
+      KAFKA_CONTROLLER_QUORUM_VOTERS: '1@kafka:29093'
+      KAFKA_PROCESS_ROLES: 'broker,controller'
+      
+      # Topic and replication settings
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
+      KAFKA_DEFAULT_REPLICATION_FACTOR: 1
+      KAFKA_NUM_PARTITIONS: 3
+      
+      # Auto-create topics
+      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true'
+      
+      # JMX for monitoring
+      KAFKA_JMX_PORT: 9997
+      KAFKA_JMX_HOSTNAME: localhost
+      
+      # Performance tuning
+      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
+      KAFKA_LOG_RETENTION_MS: 3600000  # 1 hour retention for bridge messages
+      
+      # KRaft cluster metadata
+      CLUSTER_ID: 'MkU3OEVBNTcwNTJENDM2Qk'
+    volumes:
+      - kafka_data:/var/lib/kafka/data
+    healthcheck:
+      test: ["CMD", "kafka-broker-api-versions", "--bootstrap-server", "localhost:9092"]
+      interval: 10s
+      timeout: 5s
+      retries: 5
+    networks:
+      - bridge_network
+
+  bridge1:
+    build: .
+    environment:
+      PORT: 8081
+      STORAGE_TYPE: "kafka"
+      KAFKA_BROKERS: "kafka:29092"
+      KAFKA_TOPIC: "bridge-messages"
+      KAFKA_CONSUMER_GROUP: "bridge-consumer-1"
+      CORS_ENABLE: "true"
+      HEARTBEAT_INTERVAL: 10
+      RPS_LIMIT: 1000
+      CONNECTIONS_LIMIT: 200
+    depends_on:
+      kafka:
+        condition: service_healthy
+    networks:
+      - bridge_network
+    restart: unless-stopped
+
+  bridge2:
+    build: .
+    environment:
+      PORT: 8081
+      STORAGE_TYPE: "kafka"
+      KAFKA_BROKERS: "kafka:29092"
+      KAFKA_TOPIC: "bridge-messages"
+      KAFKA_CONSUMER_GROUP: "bridge-consumer-2"
+      CORS_ENABLE: "true"
+      HEARTBEAT_INTERVAL: 10
+      RPS_LIMIT: 1000
+      CONNECTIONS_LIMIT: 200
+    depends_on:
+      kafka:
+        condition: service_healthy
+    networks:
+      - bridge_network
+    restart: unless-stopped
+
+  bridge3:
+    build: .
+    environment:
+      PORT: 8081
+      STORAGE_TYPE: "kafka"
+      KAFKA_BROKERS: "kafka:29092"
+      KAFKA_TOPIC: "bridge-messages"
+      KAFKA_CONSUMER_GROUP: "bridge-consumer-3"
+      CORS_ENABLE: "true"
+      HEARTBEAT_INTERVAL: 10
+      RPS_LIMIT: 1000
+      CONNECTIONS_LIMIT: 200
+    depends_on:
+      kafka:
+        condition: service_healthy
+    networks:
+      - bridge_network
+    restart: unless-stopped
+
+  nginx:
+    image: nginx:alpine
+    ports:
+      - "8080:80"
+    volumes:
+      - ./nginx.conf:/etc/nginx/conf.d/default.conf
+    depends_on:
+      - bridge1
+      - bridge2
+      - bridge3
+    networks:
+      - bridge_network
+    restart: unless-stopped
+
+  # Kafka UI for monitoring (now without Zookeeper dependency)
+  kafka-ui:
+    image: provectuslabs/kafka-ui:latest
+    container_name: kafka-ui
+    depends_on:
+      - kafka
+    ports:
+      - "8090:8080"
+    environment:
+      KAFKA_CLUSTERS_0_NAME: local
+      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:29092
+      # No Zookeeper configuration needed for KRaft mode
+    networks:
+      - bridge_network
+
+volumes:
+  kafka_data:
+    driver: local
+
+networks:
+  bridge_network:
+    driver: bridge
diff --git a/go.mod b/go.mod
index 774252c..93be021 100644
--- a/go.mod
+++ b/go.mod
@@ -27,14 +27,17 @@ require (
 	github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
 	github.com/jackc/pgtype v1.14.0 // indirect
 	github.com/jackc/puddle v1.3.0 // indirect
+	github.com/klauspost/compress v1.18.0 // indirect
 	github.com/labstack/gommon v0.4.2 // indirect
 	github.com/lib/pq v1.10.9 // indirect
 	github.com/mattn/go-colorable v0.1.14 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
+	github.com/pierrec/lz4/v4 v4.1.16 // indirect
 	github.com/prometheus/client_model v0.6.2 // indirect
 	github.com/prometheus/common v0.63.0 // indirect
 	github.com/prometheus/procfs v0.16.1 // indirect
+	github.com/segmentio/kafka-go v0.4.48 // indirect
 	github.com/valyala/bytebufferpool v1.0.0 // indirect
 	github.com/valyala/fasttemplate v1.2.2 // indirect
 	go.uber.org/atomic v1.11.0 // indirect
diff --git a/go.sum b/go.sum
index 716766d..3d214cb 100644
--- a/go.sum
+++ b/go.sum
@@ -101,6 +101,7 @@ github.com/jackc/puddle v1.1.3/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dv
 github.com/jackc/puddle v1.3.0 h1:eHK/5clGOatcjX3oWGBO/MpxpbHzSwud5EWTSCI+MX0=
 github.com/jackc/puddle v1.3.0/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dvjSWk=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
+github.com/klauspost/compress v1.15.9/go.mod h1:PhcZ0MbTNciWF3rruxRgKxI5NkcHHrHUDtV4Yw2GlzU=
 github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo=
 github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
@@ -144,6 +145,9 @@ github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8
 github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM=
 github.com/opencontainers/image-spec v1.1.0 h1:8SG7/vwALn54lVB/0yZ/MMwhFrPYtpEHQb2IpWsCzug=
 github.com/opencontainers/image-spec v1.1.0/go.mod h1:W4s4sFTMaBeK1BQLXbG4AdM2szdn85PY75RI83NrTrM=
+github.com/pierrec/lz4/v4 v4.1.15/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
+github.com/pierrec/lz4/v4 v4.1.16 h1:kQPfno+wyx6C5572ABwV+Uo3pDFzQ7yhyGchSyRda0c=
+github.com/pierrec/lz4/v4 v4.1.16/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
@@ -162,6 +166,8 @@ github.com/rs/xid v1.2.1/go.mod h1:+uKXf+4Djp6Md1KODXJxgGQPKngRmWyn10oCKFzNHOQ=
 github.com/rs/zerolog v1.13.0/go.mod h1:YbFCdg8HfsridGWAh22vktObvhZbQsZXe4/zB0OKkWU=
 github.com/rs/zerolog v1.15.0/go.mod h1:xYTKnLHcpfU2225ny5qZjxnj9NvkumZYjJHlAThCjNc=
 github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0=
+github.com/segmentio/kafka-go v0.4.48 h1:9jyu9CWK4W5W+SroCe8EffbrRZVqAOkuaLd/ApID4Vs=
+github.com/segmentio/kafka-go v0.4.48/go.mod h1:HjF6XbOKh0Pjlkr5GVZxt6CsjjwnmhVOfURM5KMd8qg=
 github.com/shopspring/decimal v0.0.0-20180709203117-cd690d0c9e24/go.mod h1:M+9NzErvs504Cn4c5DxATwIqPbtswREoFCre64PpcG4=
 github.com/shopspring/decimal v1.2.0 h1:abSATXmQEYyShuxI4/vyW3tV1MrKAJzCZ/0zLUXYbsQ=
 github.com/shopspring/decimal v1.2.0/go.mod h1:DKyhrW/HYNuLGql+MJL6WCR6knT2jwCFRcu2hWCYk4o=
@@ -172,17 +178,24 @@ github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVs
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.2.0/go.mod h1:qt09Ya8vawLte6SNmTgCsAVtYtaKzEcn8ATUoHMkEqE=
+github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
 github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
 github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
 github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
 github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw=
 github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
 github.com/valyala/fasttemplate v1.2.2 h1:lxLXG0uE3Qnshl9QyaK6XJxMXlQZELvChBOCmQD0Loo=
 github.com/valyala/fasttemplate v1.2.2/go.mod h1:KHLXt3tVN2HBp8eijSv/kGJopbvo7S+qRAEEKiv+SiQ=
+github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI=
+github.com/xdg-go/scram v1.1.2/go.mod h1:RT/sEzTbU5y00aCK8UOx6R7YryM0iF1N2MOmC3kKLN4=
+github.com/xdg-go/stringprep v1.0.4/go.mod h1:mPGuuIYwz7CmR2bT9j4GbQqutWS1zV24gijq1dTyGkM=
+github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
 github.com/zenazn/goji v0.9.0/go.mod h1:7S9M489iMyHBNxwZnk9/EHS098H4/F6TATF2mIxtB1Q=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.54.0 h1:TT4fX+nBOA/+LUkobKGW1ydGcn+G3vRw9+g5HwCphpk=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.54.0/go.mod h1:L7UH0GbB0p47T4Rri3uHjbpCFYrVrwc1I25QhNPiGK8=
@@ -214,6 +227,8 @@ golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPh
 golang.org/x/crypto v0.0.0-20201203163018-be400aefbc4c/go.mod h1:jdWPYTVW3xRLrWPugEBEK3UY2ZEsg3UU495nc5E+M+I=
 golang.org/x/crypto v0.0.0-20210616213533-5ff15b29337e/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.0.0-20210711020723-a769d52b0f97/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
+golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
+golang.org/x/crypto v0.14.0/go.mod h1:MVFd36DqK4CsrnJYDkBA3VC4m2GkXAM0PvzMCn4JQf4=
 golang.org/x/crypto v0.38.0 h1:jt+WWG8IZlBnVbomuhg2Mdq0+BBQaHbtqHEFEigjUV8=
 golang.org/x/crypto v0.38.0/go.mod h1:MvrbAqul58NNYPKnOra203SB9vpuZW0e+RRZV+Ggqjw=
 golang.org/x/exp v0.0.0-20230626212559-97b1e661b5df h1:UA2aFVmmsIlefxMk29Dp2juaUSth8Pyn3Tq5Y5mJGME=
@@ -221,14 +236,22 @@ golang.org/x/exp v0.0.0-20230626212559-97b1e661b5df/go.mod h1:FXUEEKJgO7OQYeo8N0
 golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
 golang.org/x/mod v0.0.0-20190513183733-4bf6d317e70e/go.mod h1:mXi4GBBbnImb6dmsKGUJ2LatrhH/nqhxcFungHvyanc=
 golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg=
+golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
+golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
 golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20190813141303-74dc4d7220e7/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
+golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
+golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
+golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg=
+golang.org/x/net v0.17.0/go.mod h1:NxSsAGuq816PNPmqtQdLE42eU2Fs7NoRIZrHJAlaCOE=
 golang.org/x/net v0.40.0 h1:79Xs7wF06Gbdcg4kdCCIQArK11Z1hr5POQ6+fIYHNuY=
 golang.org/x/net v0.40.0/go.mod h1:y0hY0exeL2Pku80/zKK7tpntoX23cqL3Oa6njdgRtds=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -241,17 +264,31 @@ golang.org/x/sys v0.0.0-20200116001909-b77594299b42/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20200223170610-d5e6a3e2c0ae/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.13.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.33.0 h1:q3i8TbbEz+JRD9ywIRlyRAQbM0qF7hu24q3teo2hbuw=
 golang.org/x/sys v0.33.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
 golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
+golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
+golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
+golang.org/x/term v0.8.0/go.mod h1:xPskH00ivmX89bAKVGSKKtLOWNx2+17Eiy94tnKShWo=
+golang.org/x/term v0.13.0/go.mod h1:LTmsnFJwVN6bCy1rVCoS+qHT1HhALEFxKncY3WNNh4U=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.4/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
+golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ=
+golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
+golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
+golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
 golang.org/x/text v0.25.0 h1:qVyWApTSYLk/drJRO5mDlNYskwQznZmkpV2c8q9zls4=
 golang.org/x/text v0.25.0/go.mod h1:WEdwpYrmk1qmdHvhkSTNPm3app7v4rsT8F2UD6+VHIA=
 golang.org/x/time v0.12.0 h1:ScB/8o8olJvc+CQPWrK3fPZNfh7qgwCrY0zJmoEQLSE=
@@ -263,7 +300,10 @@ golang.org/x/tools v0.0.0-20190621195816-6e04913cbbac/go.mod h1:/rFqwRUd4F7ZHNgw
 golang.org/x/tools v0.0.0-20190823170909-c4a336ef6a2f/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191029041327-9cc4af7d6b2c/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191029190741-b9c20aec41a5/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
+golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20200103221440-774c71fcf114/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
+golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
+golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU=
 golang.org/x/xerrors v0.0.0-20190410155217-1f06c39b4373/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20190513163551-3ee3066db522/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
diff --git a/internal/config/config.go b/internal/config/config.go
index 9ecbf28..c94baae 100644
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -9,6 +9,10 @@ import (
 var Config = struct {
 	Port                  int      `env:"PORT" envDefault:"8081"`
 	DbURI                 string   `env:"POSTGRES_URI"`
+	KafkaBrokers          []string `env:"KAFKA_BROKERS" envSeparator:","`
+	KafkaTopic            string   `env:"KAFKA_TOPIC" envDefault:"bridge-messages"`
+	KafkaConsumerGroup    string   `env:"KAFKA_CONSUMER_GROUP" envDefault:"bridge-consumer"`
+	StorageType           string   `env:"STORAGE_TYPE" envDefault:"memory"` // memory, postgres, kafka
 	WebhookURL            string   `env:"WEBHOOK_URL"`
 	CopyToURL             string   `env:"COPY_TO_URL"`
 	CorsEnable            bool     `env:"CORS_ENABLE"`
diff --git a/internal/storage/kafka.go b/internal/storage/kafka.go
new file mode 100644
index 0000000..1de5e78
--- /dev/null
+++ b/internal/storage/kafka.go
@@ -0,0 +1,292 @@
+package storage
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/callmedenchick/callmebridge/internal/models"
+	"github.com/segmentio/kafka-go"
+	log "github.com/sirupsen/logrus"
+)
+
+type KafkaMessage struct {
+	EventId   int64  `json:"event_id"`
+	Message   []byte `json:"message"`
+	ClientId  string `json:"client_id"`
+	EndTime   int64  `json:"end_time"`
+	Timestamp int64  `json:"timestamp"`
+}
+
+type cachedMessage struct {
+	models.SseMessage
+	endTime int64
+}
+
+type KafkaStorage struct {
+	brokers       []string
+	topic         string
+	consumerGroup string
+	writer        *kafka.Writer
+	reader        *kafka.Reader
+	messages      map[string][]cachedMessage // In-memory cache for quick access
+	messageMutex  sync.RWMutex
+	ctx           context.Context
+}
+
+func NewKafkaStorage(brokers []string, topic, consumerGroup string) (*KafkaStorage, error) {
+	log := log.WithField("prefix", "NewKafkaStorage")
+
+	writer := &kafka.Writer{
+		Addr:         kafka.TCP(brokers...),
+		Topic:        topic,
+		Balancer:     &kafka.LeastBytes{},
+		RequiredAcks: kafka.RequireOne,
+		Async:        false,
+	}
+
+	reader := kafka.NewReader(kafka.ReaderConfig{
+		Brokers:     brokers,
+		Topic:       topic,
+		GroupID:     consumerGroup,
+		StartOffset: kafka.LastOffset,
+		MinBytes:    1,
+		MaxBytes:    10e6, // 10MB
+	})
+
+	storage := &KafkaStorage{
+		brokers:       brokers,
+		topic:         topic,
+		consumerGroup: consumerGroup,
+		writer:        writer,
+		reader:        reader,
+		messages:      make(map[string][]cachedMessage),
+		ctx:           context.Background(),
+	}
+
+	// Start background goroutines
+	go storage.consumer()
+	go storage.cleaner()
+
+	log.Info("Kafka storage initialized successfully")
+	return storage, nil
+}
+
+func (s *KafkaStorage) Add(ctx context.Context, key string, ttl int64, mes models.SseMessage) error {
+	log := log.WithField("prefix", "KafkaStorage.Add")
+
+	kafkaMsg := KafkaMessage{
+		EventId:   mes.EventId,
+		Message:   mes.Message,
+		ClientId:  key,
+		EndTime:   time.Now().Add(time.Duration(ttl) * time.Second).Unix(),
+		Timestamp: time.Now().Unix(),
+	}
+
+	msgBytes, err := json.Marshal(kafkaMsg)
+	if err != nil {
+		log.Errorf("failed to marshal message: %v", err)
+		return err
+	}
+
+	message := kafka.Message{
+		Key:   []byte(key),
+		Value: msgBytes,
+		Time:  time.Now(),
+	}
+
+	err = s.writer.WriteMessages(ctx, message)
+	if err != nil {
+		log.Errorf("failed to write message to Kafka: %v", err)
+		return err
+	}
+
+	// Also add to in-memory cache for immediate availability
+	s.messageMutex.Lock()
+	if s.messages[key] == nil {
+		s.messages[key] = make([]cachedMessage, 0)
+	}
+	cached := cachedMessage{
+		SseMessage: mes,
+		endTime:    kafkaMsg.EndTime,
+	}
+	s.messages[key] = append(s.messages[key], cached)
+	s.messageMutex.Unlock()
+
+	log.Debugf("message added to Kafka and cache for client %s", key)
+	return nil
+}
+
+func (s *KafkaStorage) GetMessages(ctx context.Context, keys []string, lastEventId int64) ([]models.SseMessage, error) {
+	log := log.WithField("prefix", "KafkaStorage.GetMessages")
+
+	s.messageMutex.RLock()
+	defer s.messageMutex.RUnlock()
+
+	now := time.Now().Unix()
+	var result []models.SseMessage
+	for _, key := range keys {
+		if messages, exists := s.messages[key]; exists {
+			for _, msg := range messages {
+				// Skip expired messages
+				if now > msg.endTime {
+					continue
+				}
+				if msg.EventId > lastEventId {
+					result = append(result, msg.SseMessage)
+				}
+			}
+		}
+	}
+
+	log.Debugf("retrieved %d messages for clients %v", len(result), keys)
+	return result, nil
+}
+
+func (s *KafkaStorage) HealthCheck() error {
+	log := log.WithField("prefix", "KafkaStorage.HealthCheck")
+
+	// Try to create a test connection
+	conn, err := kafka.Dial("tcp", s.brokers[0])
+	if err != nil {
+		log.Errorf("kafka health check failed: %v", err)
+		return err
+	}
+	defer func() {
+		if closeErr := conn.Close(); closeErr != nil {
+			log.Errorf("failed to close connection: %v", closeErr)
+		}
+	}()
+
+	// Check if topic exists
+	partitions, err := conn.ReadPartitions(s.topic)
+	if err != nil {
+		log.Errorf("failed to read topic partitions: %v", err)
+		return err
+	}
+
+	if len(partitions) == 0 {
+		return fmt.Errorf("topic %s has no partitions", s.topic)
+	}
+
+	log.Info("kafka is healthy")
+	return nil
+}
+
+// consumer runs in background to consume messages from Kafka and maintain the in-memory cache
+func (s *KafkaStorage) consumer() {
+	log := log.WithField("prefix", "KafkaStorage.consumer")
+	log.Info("starting Kafka consumer")
+
+	for {
+		select {
+		case <-s.ctx.Done():
+			log.Info("stopping Kafka consumer")
+			return
+		default:
+			message, err := s.reader.FetchMessage(s.ctx)
+			if err != nil {
+				log.Errorf("failed to fetch message: %v", err)
+				time.Sleep(time.Second)
+				continue
+			}
+
+			var kafkaMsg KafkaMessage
+			err = json.Unmarshal(message.Value, &kafkaMsg)
+			if err != nil {
+				log.Errorf("failed to unmarshal message: %v", err)
+				if err := s.reader.CommitMessages(s.ctx, message); err != nil {
+					log.Errorf("failed to commit message: %v", err)
+				}
+				continue
+			}
+			// Check if message is still valid (not expired)
+			if time.Now().Unix() > kafkaMsg.EndTime {
+				if err := s.reader.CommitMessages(s.ctx, message); err != nil {
+					log.Errorf("failed to commit message: %v", err)
+				}
+				continue
+			}
+
+			// Add to in-memory cache
+			s.messageMutex.Lock()
+			if s.messages[kafkaMsg.ClientId] == nil {
+				s.messages[kafkaMsg.ClientId] = make([]cachedMessage, 0)
+			}
+
+			sseMsg := models.SseMessage{
+				EventId: kafkaMsg.EventId,
+				Message: kafkaMsg.Message,
+			}
+
+			cached := cachedMessage{
+				SseMessage: sseMsg,
+				endTime:    kafkaMsg.EndTime,
+			}
+
+			// Check if message already exists to avoid duplicates
+			exists := false
+			for _, existingMsg := range s.messages[kafkaMsg.ClientId] {
+				if existingMsg.EventId == cached.EventId {
+					exists = true
+					break
+				}
+			}
+
+			if !exists {
+				s.messages[kafkaMsg.ClientId] = append(s.messages[kafkaMsg.ClientId], cached)
+			}
+			s.messageMutex.Unlock()
+
+			if err := s.reader.CommitMessages(s.ctx, message); err != nil {
+				log.Errorf("failed to commit message: %v", err)
+			}
+		}
+	}
+}
+
+// cleaner runs in background to remove expired messages from the in-memory cache
+func (s *KafkaStorage) cleaner() {
+	log := log.WithField("prefix", "KafkaStorage.cleaner")
+	log.Info("starting Kafka storage cleaner")
+
+	ticker := time.NewTicker(time.Second)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-s.ctx.Done():
+			log.Info("stopping Kafka storage cleaner")
+			return
+		case <-ticker.C:
+			s.cleanExpiredMessages()
+		}
+	}
+}
+
+func (s *KafkaStorage) cleanExpiredMessages() {
+	log := log.WithField("prefix", "KafkaStorage.cleanExpiredMessages")
+
+	s.messageMutex.Lock()
+	defer s.messageMutex.Unlock()
+
+	now := time.Now().Unix()
+	for clientId, messages := range s.messages {
+		var validMessages []cachedMessage
+		for _, msg := range messages {
+			if now <= msg.endTime {
+				validMessages = append(validMessages, msg)
+			}
+		}
+
+		if len(validMessages) == 0 {
+			delete(s.messages, clientId)
+		} else {
+			s.messages[clientId] = validMessages
+		}
+	}
+
+	log.Debug("expired messages cleaned from cache")
+}
diff --git a/internal/storage/storage.go b/internal/storage/storage.go
index 5281467..971318b 100644
--- a/internal/storage/storage.go
+++ b/internal/storage/storage.go
@@ -2,6 +2,8 @@ package storage
 
 import (
 	"context"
+	"fmt"
+	"strings"
 
 	"github.com/callmedenchick/callmebridge/internal/models"
 )
@@ -12,7 +14,42 @@ type Storage interface {
 	HealthCheck() error
 }
 
-func NewStorage(dbURI string) (Storage, error) {
+// StorageConfig holds configuration for different storage types
+type StorageConfig struct {
+	Type               string
+	PostgresURI        string
+	KafkaBrokers       []string
+	KafkaTopic         string
+	KafkaConsumerGroup string
+}
+
+func NewStorage(config StorageConfig) (Storage, error) {
+	switch strings.ToLower(config.Type) {
+	case "postgres", "pg":
+		if config.PostgresURI == "" {
+			return nil, fmt.Errorf("postgres URI is required for postgres storage")
+		}
+		return NewPgStorage(config.PostgresURI)
+	case "kafka":
+		if len(config.KafkaBrokers) == 0 {
+			return nil, fmt.Errorf("kafka brokers are required for kafka storage")
+		}
+		if config.KafkaTopic == "" {
+			config.KafkaTopic = "bridge-messages"
+		}
+		if config.KafkaConsumerGroup == "" {
+			config.KafkaConsumerGroup = "bridge-consumer"
+		}
+		return NewKafkaStorage(config.KafkaBrokers, config.KafkaTopic, config.KafkaConsumerGroup)
+	case "memory", "mem", "":
+		return NewMemStorage(), nil
+	default:
+		return nil, fmt.Errorf("unsupported storage type: %s", config.Type)
+	}
+}
+
+// Deprecated: Use NewStorage with StorageConfig instead
+func NewStorageDeprecated(dbURI string) (Storage, error) {
 	if dbURI != "" {
 		return NewPgStorage(dbURI)
 	}
diff --git a/recipes/docker-compose-kafka.yml b/recipes/docker-compose-kafka.yml
new file mode 100644
index 0000000..24aef5e
--- /dev/null
+++ b/recipes/docker-compose-kafka.yml
@@ -0,0 +1,50 @@
+services:
+  kafka:
+    image: confluentinc/cp-kafka:7.4.0
+    container_name: kafka
+    ports:
+      - "9092:9092"
+    environment:
+      # KRaft mode configuration
+      KAFKA_NODE_ID: 1
+      KAFKA_PROCESS_ROLES: 'broker,controller'
+      KAFKA_CONTROLLER_QUORUM_VOTERS: '1@localhost:29093'
+      KAFKA_LISTENERS: 'PLAINTEXT://localhost:29092,CONTROLLER://localhost:29093,PLAINTEXT_HOST://0.0.0.0:9092'
+      KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT'
+      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://localhost:29092,PLAINTEXT_HOST://localhost:9092'
+      KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT'
+      CLUSTER_ID: 'ciWo7IWazngRchmPES6q5A=='
+      
+      # Basic settings
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
+      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
+      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true'
+      
+      # Memory optimization
+      KAFKA_HEAP_OPTS: "-Xmx512M -Xms256M"
+    deploy:
+      resources:
+        limits:
+          memory: 1G
+        reservations:
+          memory: 512M
+    healthcheck:
+      test: ["CMD", "kafka-broker-api-versions", "--bootstrap-server", "localhost:9092"]
+      interval: 30s
+      timeout: 10s
+      retries: 5
+      start_period: 60s
+
+  kafka-ui:
+    image: provectuslabs/kafka-ui:latest
+    container_name: kafka-ui
+    depends_on:
+      - kafka
+    ports:
+      - "8080:8080"
+    environment:
+      KAFKA_CLUSTERS_0_NAME: local
+      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: localhost:9092
\ No newline at end of file
diff --git a/recipes/kafka.md b/recipes/kafka.md
new file mode 100644
index 0000000..ad1cb16
--- /dev/null
+++ b/recipes/kafka.md
@@ -0,0 +1,48 @@
+
+````yaml
+
+````
+
+## Usage
+
+````bash
+# Start everything
+docker-compose -f docker-compose-kafka.yml up -d
+
+# Check logs
+docker-compose -f docker-compose-kafka.yml logs -f kafka
+docker-compose -f docker-compose-kafka.yml logs -f kafka-ui
+
+# Stop everything
+docker-compose -f docker-compose-kafka.yml down
+
+# Stop and remove volumes (clean slate)
+docker-compose -f docker-compose-kafka.yml down -v
+````
+
+## Access
+
+- **Kafka**: `localhost:9092`
+- **Kafka UI**: `http://localhost:8080`
+
+## Test It
+
+````bash
+# Wait for services to be healthy, then create a topic
+docker exec kafka kafka-topics --create --topic sse-messages --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1
+
+# Send a test message
+echo "Hello from Docker Kafka!" | docker exec -i kafka kafka-console-producer --topic sse-messages --bootstrap-server localhost:9092
+````
+
+```
+kafka-topics \
+  --create \
+  --topic bridge-messages \
+  --bootstrap-server localhost:9092 \
+  --partitions 1000 \
+  --replication-factor 1 \
+  --config retention.ms=600000 \
+  --config segment.ms=300000 \
+  --config cleanup.policy=delete
+```
\ No newline at end of file
diff --git a/scripts/integration-test.sh b/scripts/integration-test.sh
index e4a5b15..42ba5ea 100755
--- a/scripts/integration-test.sh
+++ b/scripts/integration-test.sh
@@ -1,39 +1,161 @@
 #!/bin/bash
 
-# Script to run integration tests with bridge-sdk
+# Enhanced integration test script supporting multiple storage types
+# Usage: ./integration-test.sh [storage_type]
+# storage_type: memory (default), postgres, kafka
+
 set -e
 
-BRIDGE_PORT=${PORT:-8081}
+# Configuration
+STORAGE_TYPE=${1:-memory}
+BRIDGE_PORT=${PORT:-8082}  # Use different port to avoid conflicts
 BRIDGE_URL=${BRIDGE_URL:-http://localhost:$BRIDGE_PORT}
 BRIDGE_BINARY="./callmebridge"
 
-echo "🚀 Starting integration tests with bridge-sdk..."
-echo "Bridge URL: $BRIDGE_URL"
-
-# Build the bridge if it doesn't exist
-if [ ! -f "$BRIDGE_BINARY" ]; then
-    echo "🔨 Building bridge..."
-    make build
-fi
+# Storage-specific configuration
+POSTGRES_CONTAINER=""
+KAFKA_CONTAINER=""
+TEST_TIMEOUT=60
 
-# Start the bridge in background
-echo "🌉 Starting bridge server on port $BRIDGE_PORT..."
-$BRIDGE_BINARY &
-BRIDGE_PID=$!
+echo "🚀 Starting integration tests with $STORAGE_TYPE storage..."
+echo "Bridge URL: $BRIDGE_URL"
 
 # Function to cleanup
 cleanup() {
     echo "🧹 Cleaning up..."
-    if kill -0 $BRIDGE_PID 2>/dev/null; then
-        echo "🛑 Stopping bridge server (PID: $BRIDGE_PID)..."
+    
+    # Stop bridge if running
+    if [ ! -z "$BRIDGE_PID" ] && kill -0 $BRIDGE_PID 2>/dev/null; then
+        echo "� Stopping bridge server (PID: $BRIDGE_PID)..."
         kill $BRIDGE_PID
         wait $BRIDGE_PID 2>/dev/null || true
     fi
+    
+    # Clean up storage-specific resources
+    case $STORAGE_TYPE in
+        postgres)
+            if [ ! -z "$POSTGRES_CONTAINER" ]; then
+                echo "🗄️ Stopping PostgreSQL container..."
+                docker stop $POSTGRES_CONTAINER >/dev/null 2>&1 || true
+                docker rm $POSTGRES_CONTAINER >/dev/null 2>&1 || true
+            fi
+            ;;
+        kafka)
+            if [ ! -z "$KAFKA_CONTAINER" ]; then
+                echo "📨 Stopping Kafka container..."
+                docker stop $KAFKA_CONTAINER >/dev/null 2>&1 || true
+                docker rm $KAFKA_CONTAINER >/dev/null 2>&1 || true
+            fi
+            ;;
+    esac
 }
 
 # Set trap to cleanup on exit
 trap cleanup EXIT
 
+# Function to setup storage
+setup_storage() {
+    case $STORAGE_TYPE in
+        memory)
+            echo "💾 Using in-memory storage (no setup required)"
+            export STORAGE_TYPE="memory"
+            ;;
+        postgres)
+            echo "🗄️ Setting up PostgreSQL storage..."
+            
+            # Start PostgreSQL container
+            POSTGRES_CONTAINER=$(docker run -d \
+                --name "integration-postgres-$$" \
+                -e POSTGRES_DB=bridge \
+                -e POSTGRES_USER=bridge_user \
+                -e POSTGRES_PASSWORD=bridge_password \
+                -p 5433:5432 \
+                postgres:15-alpine)
+            
+            echo "⏳ Waiting for PostgreSQL to be ready..."
+            for i in {1..30}; do
+                if docker exec $POSTGRES_CONTAINER pg_isready -U bridge_user -d bridge >/dev/null 2>&1; then
+                    echo "✅ PostgreSQL is ready!"
+                    break
+                fi
+                if [ $i -eq 30 ]; then
+                    echo "❌ PostgreSQL failed to start within 30 seconds"
+                    exit 1
+                fi
+                sleep 1
+            done
+            
+            export STORAGE_TYPE="postgres"
+            export POSTGRES_URI="postgres://bridge_user:bridge_password@localhost:5433/bridge?sslmode=disable"
+            ;;
+        kafka)
+            echo "📨 Setting up Kafka storage..."
+            
+            # Start Kafka container in KRaft mode
+            KAFKA_CONTAINER=$(docker run -d \
+                --name "integration-kafka-$$" \
+                -p 9094:9092 \
+                -e KAFKA_NODE_ID=1 \
+                -e KAFKA_LISTENER_SECURITY_PROTOCOL_MAP='CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT' \
+                -e KAFKA_ADVERTISED_LISTENERS='PLAINTEXT://localhost:29092,PLAINTEXT_HOST://localhost:9094' \
+                -e KAFKA_LISTENERS='PLAINTEXT://localhost:29092,CONTROLLER://localhost:29093,PLAINTEXT_HOST://0.0.0.0:9092' \
+                -e KAFKA_INTER_BROKER_LISTENER_NAME='PLAINTEXT' \
+                -e KAFKA_CONTROLLER_LISTENER_NAMES='CONTROLLER' \
+                -e KAFKA_CONTROLLER_QUORUM_VOTERS='1@localhost:29093' \
+                -e KAFKA_PROCESS_ROLES='broker,controller' \
+                -e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \
+                -e KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=1 \
+                -e KAFKA_TRANSACTION_STATE_LOG_MIN_ISR=1 \
+                -e KAFKA_DEFAULT_REPLICATION_FACTOR=1 \
+                -e KAFKA_AUTO_CREATE_TOPICS_ENABLE='true' \
+                -e KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS=0 \
+                -e CLUSTER_ID='MkU3OEVBNTcwNTJENDM2Qk' \
+                confluentinc/cp-kafka:7.4.0)
+            
+            echo "⏳ Waiting for Kafka to be ready..."
+            for i in {1..60}; do
+                if docker exec $KAFKA_CONTAINER kafka-broker-api-versions --bootstrap-server localhost:9092 >/dev/null 2>&1; then
+                    echo "✅ Kafka is ready!"
+                    break
+                fi
+                if [ $i -eq 60 ]; then
+                    echo "❌ Kafka failed to start within 60 seconds"
+                    exit 1
+                fi
+                sleep 1
+            done
+            
+            export STORAGE_TYPE="kafka"
+            export KAFKA_BROKERS="localhost:9094"
+            export KAFKA_TOPIC="bridge-integration-test"
+            export KAFKA_CONSUMER_GROUP="bridge-integration-consumer"
+            ;;
+        *)
+            echo "❌ Unknown storage type: $STORAGE_TYPE"
+            echo "Supported types: memory, postgres, kafka"
+            exit 1
+            ;;
+    esac
+}
+
+# Build the bridge if it doesn't exist
+if [ ! -f "$BRIDGE_BINARY" ]; then
+    echo "🔨 Building bridge..."
+    make build
+fi
+
+# Setup storage
+setup_storage
+
+# Start the bridge in background
+echo "🌉 Starting bridge server on port $BRIDGE_PORT with $STORAGE_TYPE storage..."
+export PORT=$BRIDGE_PORT
+export CORS_ENABLE=true
+export HEARTBEAT_INTERVAL=5
+
+$BRIDGE_BINARY &
+BRIDGE_PID=$!
+
 # Wait for bridge to be ready
 echo "⏳ Waiting for bridge to be ready..."
 for i in {1..30}; do
@@ -45,12 +167,77 @@ for i in {1..30}; do
         echo "❌ Bridge failed to start within 30 seconds"
         exit 1
     fi
-        echo "⏳ Waiting for bridge to be ready... ($i/30)"
+    echo "⏳ Waiting for bridge to be ready... ($i/30)"
     sleep 1
 done
 
+# Check readiness (including storage health)
+echo "🏥 Checking bridge readiness (including storage health)..."
+if ! curl -f "$BRIDGE_URL/ready" &> /dev/null; then
+    echo "❌ Bridge readiness check failed"
+    exit 1
+fi
+echo "✅ Bridge readiness check passed!"
+
 # Run bridge-sdk tests
-echo "🧪 Running bridge-sdk tests against live bridge..."
-BRIDGE_URL="$BRIDGE_URL/bridge" ./scripts/test-bridge-sdk.sh
+echo "🧪 Running bridge-sdk tests against live bridge with $STORAGE_TYPE storage..."
+# Use timeout if available, otherwise run without it
+if command -v timeout >/dev/null 2>&1; then
+    BRIDGE_URL="$BRIDGE_URL/bridge" timeout $TEST_TIMEOUT ./scripts/test-bridge-sdk.sh
+else
+    BRIDGE_URL="$BRIDGE_URL/bridge" ./scripts/test-bridge-sdk.sh
+fi
+
+# Run storage-specific tests
+echo "🔍 Running storage-specific integration tests..."
+case $STORAGE_TYPE in
+    kafka)
+        echo "📨 Testing Kafka-specific functionality..."
+        
+        # Test multiple consumers
+        CLIENT_A="integration-client-a"
+        CLIENT_B="integration-client-b"
+        
+        # Start SSE connections
+        curl -N -H "Accept: text/event-stream" "$BRIDGE_URL/events?client_id=$CLIENT_A" > /tmp/client_a_output.txt &
+        SSE_PID_A=$!
+        
+        curl -N -H "Accept: text/event-stream" "$BRIDGE_URL/events?client_id=$CLIENT_B" > /tmp/client_b_output.txt &
+        SSE_PID_B=$!
+        
+        sleep 2  # Allow connections to establish
+        
+        # Send messages
+        curl -X POST "$BRIDGE_URL/message?client_id=sender&to=$CLIENT_A&ttl=60" -d "Test message A" >/dev/null 2>&1
+        curl -X POST "$BRIDGE_URL/message?client_id=sender&to=$CLIENT_B&ttl=60" -d "Test message B" >/dev/null 2>&1
+        
+        sleep 3  # Allow message processing
+        
+        # Check if messages were received
+        kill $SSE_PID_A $SSE_PID_B 2>/dev/null || true
+        wait $SSE_PID_A $SSE_PID_B 2>/dev/null || true
+        
+        if grep -q "Test message A" /tmp/client_a_output.txt && grep -q "Test message B" /tmp/client_b_output.txt; then
+            echo "✅ Kafka multi-client test passed!"
+        else
+            echo "❌ Kafka multi-client test failed!"
+            echo "Client A output:"
+            cat /tmp/client_a_output.txt 2>/dev/null || echo "No output"
+            echo "Client B output:"
+            cat /tmp/client_b_output.txt 2>/dev/null || echo "No output"
+            exit 1
+        fi
+        
+        rm -f /tmp/client_a_output.txt /tmp/client_b_output.txt
+        ;;
+    postgres)
+        echo "🗄️ Testing PostgreSQL-specific functionality..."
+        # Could add database-specific tests here
+        ;;
+    memory)
+        echo "💾 Testing memory storage functionality..."
+        # Memory storage tests (if any specific tests needed)
+        ;;
+esac
 
-echo "🎉 Integration tests completed successfully!"
+echo "🎉 Integration tests for $STORAGE_TYPE storage completed successfully!"
diff --git a/scripts/run-with-kafka.sh b/scripts/run-with-kafka.sh
new file mode 100755
index 0000000..cf7e920
--- /dev/null
+++ b/scripts/run-with-kafka.sh
@@ -0,0 +1,69 @@
+#!/bin/bash
+
+# Example script to run CallMeBridge with Kafka storage
+
+echo "🚀 Starting CallMeBridge with Kafka storage"
+echo ""
+
+# Check if Kafka is running
+echo "📋 Checking Kafka availability..."
+if ! nc -z localhost 9092; then
+    echo "❌ Kafka is not running on localhost:9092"
+    echo "💡 Please start Kafka first:"
+    echo "   Option 1: Use docker-compose (KRaft mode - no Zookeeper needed!):"
+    echo "     docker-compose -f docker-compose.kafka.yml up -d kafka"
+    echo ""
+    echo "   Option 2: Use local Kafka installation with KRaft:"
+    echo "     # Generate cluster ID"
+    echo "     KAFKA_CLUSTER_ID=\$(kafka-storage.sh random-uuid)"
+    echo "     # Format log directories"
+    echo "     kafka-storage.sh format -t \$KAFKA_CLUSTER_ID -c /usr/local/etc/kafka/kraft/server.properties"
+    echo "     # Start Kafka in KRaft mode"
+    echo "     kafka-server-start.sh /usr/local/etc/kafka/kraft/server.properties"
+    echo ""
+    exit 1
+fi
+
+echo "✅ Kafka is running"
+echo ""
+
+# Set environment variables for Kafka storage
+export STORAGE_TYPE="kafka"
+export KAFKA_BROKERS="localhost:9092"
+export KAFKA_TOPIC="bridge-messages"
+export KAFKA_CONSUMER_GROUP="bridge-consumer"
+export PORT="8081"
+export CORS_ENABLE="true"
+export HEARTBEAT_INTERVAL="10"
+export RPS_LIMIT="1000"
+export CONNECTIONS_LIMIT="200"
+
+echo "🔧 Configuration:"
+echo "   Storage Type: $STORAGE_TYPE"
+echo "   Kafka Brokers: $KAFKA_BROKERS"
+echo "   Kafka Topic: $KAFKA_TOPIC"
+echo "   Consumer Group: $KAFKA_CONSUMER_GROUP"
+echo "   Port: $PORT"
+echo ""
+
+# Build and run the bridge
+echo "🔨 Building CallMeBridge..."
+if ! make build; then
+    echo "❌ Failed to build CallMeBridge"
+    exit 1
+fi
+
+echo "✅ Build successful"
+echo ""
+
+echo "🌉 Starting CallMeBridge with Kafka storage..."
+echo "   API endpoints will be available at:"
+echo "   - http://localhost:8081/bridge/events (SSE)"
+echo "   - http://localhost:8081/bridge/message (POST)"
+echo "   - http://localhost:8081/health (Health check)"
+echo "   - http://localhost:8081/ready (Readiness check)"
+echo "   - http://localhost:8081/metrics (Prometheus metrics)"
+echo ""
+
+# Run the bridge
+./callmebridge
diff --git a/scripts/test-kafka-integration.sh b/scripts/test-kafka-integration.sh
new file mode 100755
index 0000000..0cf2ed5
--- /dev/null
+++ b/scripts/test-kafka-integration.sh
@@ -0,0 +1,150 @@
+#!/bin/bash
+
+# Integration test for CallMeBridge with Kafka storage
+
+set -e
+
+BRIDGE_URL="http://localhost:8081"
+CLIENT_ID="test-client-$(date +%s)"
+MESSAGE_CONTENT="Hello from Kafka integration test!"
+
+echo "🧪 CallMeBridge Kafka Integration Test"
+echo "======================================"
+echo ""
+
+# Function to cleanup background processes
+cleanup() {
+    echo "🧹 Cleaning up..."
+    if [ ! -z "$BRIDGE_PID" ]; then
+        kill $BRIDGE_PID 2>/dev/null || true
+    fi
+    if [ ! -z "$SSE_PID" ]; then
+        kill $SSE_PID 2>/dev/null || true
+    fi
+}
+
+trap cleanup EXIT
+
+# Check if Kafka is running
+echo "📋 Checking Kafka availability..."
+if ! nc -z localhost 9092; then
+    echo "❌ Kafka is not running on localhost:9092"
+    echo "Please start Kafka with: docker-compose -f docker-compose.kafka.yml up -d kafka"
+    exit 1
+fi
+echo "✅ Kafka is available"
+
+# Check if bridge is built
+echo "🔨 Building CallMeBridge..."
+make build
+
+# Start bridge with Kafka storage
+echo "🌉 Starting CallMeBridge with Kafka storage..."
+export STORAGE_TYPE="kafka"
+export KAFKA_BROKERS="localhost:9092"
+export KAFKA_TOPIC="bridge-test-messages"
+export KAFKA_CONSUMER_GROUP="bridge-test-consumer"
+export PORT="8081"
+export CORS_ENABLE="true"
+
+./callmebridge > bridge.log 2>&1 &
+BRIDGE_PID=$!
+
+# Wait for bridge to start
+echo "⏳ Waiting for bridge to start..."
+for i in {1..30}; do
+    if curl -s "$BRIDGE_URL/health" > /dev/null 2>&1; then
+        break
+    fi
+    sleep 1
+    if [ $i -eq 30 ]; then
+        echo "❌ Bridge failed to start within 30 seconds"
+        echo "Bridge logs:"
+        cat bridge.log
+        exit 1
+    fi
+done
+echo "✅ Bridge is running"
+
+# Test health endpoint
+echo "🏥 Testing health endpoint..."
+HEALTH_RESPONSE=$(curl -s "$BRIDGE_URL/health")
+if echo "$HEALTH_RESPONSE" | grep -q '"status":"ok"'; then
+    echo "✅ Health check passed"
+else
+    echo "❌ Health check failed: $HEALTH_RESPONSE"
+    exit 1
+fi
+
+# Test ready endpoint
+echo "🚦 Testing ready endpoint..."
+READY_RESPONSE=$(curl -s "$BRIDGE_URL/ready")
+if echo "$READY_RESPONSE" | grep -q '"status":"ready"'; then
+    echo "✅ Ready check passed"
+else
+    echo "❌ Ready check failed: $READY_RESPONSE"
+    exit 1
+fi
+
+# Start SSE connection in background
+echo "📡 Starting SSE connection..."
+curl -s -N "$BRIDGE_URL/bridge/events?client_id=$CLIENT_ID" > sse_output.txt &
+SSE_PID=$!
+
+# Wait a moment for SSE connection to establish
+sleep 2
+
+# Send a test message
+echo "📤 Sending test message..."
+SEND_RESPONSE=$(curl -s -X POST "$BRIDGE_URL/bridge/message" \
+    -H "Content-Type: application/json" \
+    -d "{\"from\":\"$CLIENT_ID\",\"message\":\"$MESSAGE_CONTENT\"}")
+
+if echo "$SEND_RESPONSE" | grep -q '"status":"ok"'; then
+    echo "✅ Message sent successfully"
+else
+    echo "❌ Message send failed: $SEND_RESPONSE"
+    exit 1
+fi
+
+# Wait for message to be processed and delivered
+echo "⏳ Waiting for message delivery..."
+sleep 5
+
+# Check if message was received via SSE
+echo "📥 Checking received messages..."
+if [ -f sse_output.txt ] && grep -q "$MESSAGE_CONTENT" sse_output.txt; then
+    echo "✅ Message received via SSE"
+    echo "📋 Received content:"
+    grep "$MESSAGE_CONTENT" sse_output.txt | head -1
+else
+    echo "❌ Message not received via SSE"
+    echo "SSE output:"
+    cat sse_output.txt 2>/dev/null || echo "No SSE output file"
+    exit 1
+fi
+
+# Test metrics endpoint
+echo "📊 Testing metrics endpoint..."
+METRICS_RESPONSE=$(curl -s "$BRIDGE_URL/metrics")
+if echo "$METRICS_RESPONSE" | grep -q "bridge_health_status"; then
+    echo "✅ Metrics endpoint working"
+else
+    echo "❌ Metrics endpoint failed"
+    exit 1
+fi
+
+echo ""
+echo "🎉 All tests passed!"
+echo "✅ Kafka storage integration is working correctly"
+echo ""
+
+# Show some metrics
+echo "📈 Bridge metrics:"
+echo "$(curl -s "$BRIDGE_URL/metrics" | grep -E "(bridge_health_status|bridge_ready_status)" | head -2)"
+
+# Clean up test files
+rm -f sse_output.txt bridge.log
+
+echo ""
+echo "🧪 Integration test completed successfully!"