enhance: Update to the latest CipherPlugin API (#41599)

See also: #41585

Signed-off-by: yangxuan <xuan.yang@zilliz.com>

go.mod

@@ -21,7 +21,7 @@ require (
 	github.com/grpc-ecosystem/go-grpc-middleware v1.3.0
 	github.com/klauspost/compress v1.17.9
 	github.com/mgutz/ansi v0.0.0-20200706080929-d51e80ef957d
-	github.com/milvus-io/milvus-proto/go-api/v2 v2.5.0-beta.0.20250407030015-dcf7688ad54a
+	github.com/milvus-io/milvus-proto/go-api/v2 v2.5.0-beta.0.20250429023443-20f3b3d3e030
 	github.com/minio/minio-go/v7 v7.0.73
 	github.com/pingcap/log v1.1.1-0.20221015072633-39906604fb81
 	github.com/prometheus/client_golang v1.14.0

go.sum

@@ -740,6 +740,8 @@ github.com/milvus-io/gorocksdb v0.0.0-20220624081344-8c5f4212846b h1:TfeY0NxYxZz
 github.com/milvus-io/gorocksdb v0.0.0-20220624081344-8c5f4212846b/go.mod h1:iwW+9cWfIzzDseEBCCeDSN5SD16Tidvy8cwQ7ZY8Qj4=
 github.com/milvus-io/milvus-proto/go-api/v2 v2.5.0-beta.0.20250407030015-dcf7688ad54a h1:W+9nVXKcI9FdiyrFbrs9BIFUqRW0pLY+Fn0fsmmuLyw=
 github.com/milvus-io/milvus-proto/go-api/v2 v2.5.0-beta.0.20250407030015-dcf7688ad54a/go.mod h1:/6UT4zZl6awVeXLeE7UGDWZvXj3IWkRsh3mqsn0DiAs=
+github.com/milvus-io/milvus-proto/go-api/v2 v2.5.0-beta.0.20250429023443-20f3b3d3e030 h1:TZR8QUxLExFI/LBexz2/ucWGvvDZv73IXb/dUNlwRrg=
+github.com/milvus-io/milvus-proto/go-api/v2 v2.5.0-beta.0.20250429023443-20f3b3d3e030/go.mod h1:/6UT4zZl6awVeXLeE7UGDWZvXj3IWkRsh3mqsn0DiAs=
 github.com/milvus-io/pulsar-client-go v0.12.1 h1:O2JZp1tsYiO7C0MQ4hrUY/aJXnn2Gry6hpm7UodghmE=
 github.com/milvus-io/pulsar-client-go v0.12.1/go.mod h1:dkutuH4oS2pXiGm+Ti7fQZ4MRjrMPZ8IJeEGAWMeckk=
 github.com/minio/asm2plan9s v0.0.0-20200509001527-cdd76441f9d8 h1:AMFGa4R4MiIpspGNG7Z948v4n35fFGB3RR3G/ry4FWs=

pkg/go.mod

@@ -20,7 +20,7 @@ require (
 	github.com/jolestar/go-commons-pool/v2 v2.1.2
 	github.com/json-iterator/go v1.1.12
 	github.com/klauspost/compress v1.17.9
-	github.com/milvus-io/milvus-proto/go-api/v2 v2.5.0-beta.0.20250325034212-6e98baa34971
+	github.com/milvus-io/milvus-proto/go-api/v2 v2.5.0-beta.0.20250429023443-20f3b3d3e030
 	github.com/minio/minio-go/v7 v7.0.73
 	github.com/panjf2000/ants/v2 v2.7.2
 	github.com/prometheus/client_golang v1.14.0

pkg/go.sum

@@ -560,6 +560,8 @@ github.com/milvus-io/gorocksdb v0.0.0-20220624081344-8c5f4212846b h1:TfeY0NxYxZz
 github.com/milvus-io/gorocksdb v0.0.0-20220624081344-8c5f4212846b/go.mod h1:iwW+9cWfIzzDseEBCCeDSN5SD16Tidvy8cwQ7ZY8Qj4=
 github.com/milvus-io/milvus-proto/go-api/v2 v2.5.0-beta.0.20250325034212-6e98baa34971 h1:CKKrOtri+dbTUkMJehDuSM489OIqJab1t0pUq+PV73E=
 github.com/milvus-io/milvus-proto/go-api/v2 v2.5.0-beta.0.20250325034212-6e98baa34971/go.mod h1:/6UT4zZl6awVeXLeE7UGDWZvXj3IWkRsh3mqsn0DiAs=
+github.com/milvus-io/milvus-proto/go-api/v2 v2.5.0-beta.0.20250429023443-20f3b3d3e030 h1:TZR8QUxLExFI/LBexz2/ucWGvvDZv73IXb/dUNlwRrg=
+github.com/milvus-io/milvus-proto/go-api/v2 v2.5.0-beta.0.20250429023443-20f3b3d3e030/go.mod h1:/6UT4zZl6awVeXLeE7UGDWZvXj3IWkRsh3mqsn0DiAs=
 github.com/milvus-io/pulsar-client-go v0.12.1 h1:O2JZp1tsYiO7C0MQ4hrUY/aJXnn2Gry6hpm7UodghmE=
 github.com/milvus-io/pulsar-client-go v0.12.1/go.mod h1:dkutuH4oS2pXiGm+Ti7fQZ4MRjrMPZ8IJeEGAWMeckk=
 github.com/minio/md5-simd v1.1.2 h1:Gdi1DZK69+ZVMoNHRXJyNcxrMA4dSxoYHZSQbirFg34=

pkg/mocks/github.com/milvus-io/milvus-proto/go-api/v2/mock_hook/mock_Cipher.go

@@ -20,9 +20,9 @@ func (_m *MockCipher) EXPECT() *MockCipher_Expecter {
 	return &MockCipher_Expecter{mock: &_m.Mock}
 }
 
-// GetDecryptor provides a mock function with given fields: ezID, safeKey
+// GetDecryptor provides a mock function with given fields: ezID, collectionID, safeKey
-func (_m *MockCipher) GetDecryptor(ezID int64, safeKey []byte) (hook.Decryptor, error) {
+func (_m *MockCipher) GetDecryptor(ezID int64, collectionID int64, safeKey []byte) (hook.Decryptor, error) {
-	ret := _m.Called(ezID, safeKey)
+	ret := _m.Called(ezID, collectionID, safeKey)
 
 	if len(ret) == 0 {
 		panic("no return value specified for GetDecryptor")
@@ -30,19 +30,19 @@ func (_m *MockCipher) GetDecryptor(ezID int64, safeKey []byte) (hook.Decryptor,
 
 	var r0 hook.Decryptor
 	var r1 error
-	if rf, ok := ret.Get(0).(func(int64, []byte) (hook.Decryptor, error)); ok {
+	if rf, ok := ret.Get(0).(func(int64, int64, []byte) (hook.Decryptor, error)); ok {
-		return rf(ezID, safeKey)
+		return rf(ezID, collectionID, safeKey)
 	}
-	if rf, ok := ret.Get(0).(func(int64, []byte) hook.Decryptor); ok {
+	if rf, ok := ret.Get(0).(func(int64, int64, []byte) hook.Decryptor); ok {
-		r0 = rf(ezID, safeKey)
+		r0 = rf(ezID, collectionID, safeKey)
 	} else {
 		if ret.Get(0) != nil {
 			r0 = ret.Get(0).(hook.Decryptor)
 		}
 	}
 
-	if rf, ok := ret.Get(1).(func(int64, []byte) error); ok {
+	if rf, ok := ret.Get(1).(func(int64, int64, []byte) error); ok {
-		r1 = rf(ezID, safeKey)
+		r1 = rf(ezID, collectionID, safeKey)
 	} else {
 		r1 = ret.Error(1)
 	}
@@ -57,14 +57,15 @@ type MockCipher_GetDecryptor_Call struct {
 
 // GetDecryptor is a helper method to define mock.On call
 //   - ezID int64
+//   - collectionID int64
 //   - safeKey []byte
-func (_e *MockCipher_Expecter) GetDecryptor(ezID interface{}, safeKey interface{}) *MockCipher_GetDecryptor_Call {
+func (_e *MockCipher_Expecter) GetDecryptor(ezID interface{}, collectionID interface{}, safeKey interface{}) *MockCipher_GetDecryptor_Call {
-	return &MockCipher_GetDecryptor_Call{Call: _e.mock.On("GetDecryptor", ezID, safeKey)}
+	return &MockCipher_GetDecryptor_Call{Call: _e.mock.On("GetDecryptor", ezID, collectionID, safeKey)}
 }
 
-func (_c *MockCipher_GetDecryptor_Call) Run(run func(ezID int64, safeKey []byte)) *MockCipher_GetDecryptor_Call {
+func (_c *MockCipher_GetDecryptor_Call) Run(run func(ezID int64, collectionID int64, safeKey []byte)) *MockCipher_GetDecryptor_Call {
 	_c.Call.Run(func(args mock.Arguments) {
-		run(args[0].(int64), args[1].([]byte))
+		run(args[0].(int64), args[1].(int64), args[2].([]byte))
 	})
 	return _c
 }
@@ -74,14 +75,14 @@ func (_c *MockCipher_GetDecryptor_Call) Return(_a0 hook.Decryptor, _a1 error) *M
 	return _c
 }
 
-func (_c *MockCipher_GetDecryptor_Call) RunAndReturn(run func(int64, []byte) (hook.Decryptor, error)) *MockCipher_GetDecryptor_Call {
+func (_c *MockCipher_GetDecryptor_Call) RunAndReturn(run func(int64, int64, []byte) (hook.Decryptor, error)) *MockCipher_GetDecryptor_Call {
 	_c.Call.Return(run)
 	return _c
 }
 
-// GetEncryptor provides a mock function with given fields: ezID
+// GetEncryptor provides a mock function with given fields: ezID, collectionID
-func (_m *MockCipher) GetEncryptor(ezID int64) (hook.Encryptor, []byte, error) {
+func (_m *MockCipher) GetEncryptor(ezID int64, collectionID int64) (hook.Encryptor, []byte, error) {
-	ret := _m.Called(ezID)
+	ret := _m.Called(ezID, collectionID)
 
 	if len(ret) == 0 {
 		panic("no return value specified for GetEncryptor")
@@ -90,27 +91,27 @@ func (_m *MockCipher) GetEncryptor(ezID int64) (hook.Encryptor, []byte, error) {
 	var r0 hook.Encryptor
 	var r1 []byte
 	var r2 error
-	if rf, ok := ret.Get(0).(func(int64) (hook.Encryptor, []byte, error)); ok {
+	if rf, ok := ret.Get(0).(func(int64, int64) (hook.Encryptor, []byte, error)); ok {
-		return rf(ezID)
+		return rf(ezID, collectionID)
 	}
-	if rf, ok := ret.Get(0).(func(int64) hook.Encryptor); ok {
+	if rf, ok := ret.Get(0).(func(int64, int64) hook.Encryptor); ok {
-		r0 = rf(ezID)
+		r0 = rf(ezID, collectionID)
 	} else {
 		if ret.Get(0) != nil {
 			r0 = ret.Get(0).(hook.Encryptor)
 		}
 	}
 
-	if rf, ok := ret.Get(1).(func(int64) []byte); ok {
+	if rf, ok := ret.Get(1).(func(int64, int64) []byte); ok {
-		r1 = rf(ezID)
+		r1 = rf(ezID, collectionID)
 	} else {
 		if ret.Get(1) != nil {
 			r1 = ret.Get(1).([]byte)
 		}
 	}
 
-	if rf, ok := ret.Get(2).(func(int64) error); ok {
+	if rf, ok := ret.Get(2).(func(int64, int64) error); ok {
-		r2 = rf(ezID)
+		r2 = rf(ezID, collectionID)
 	} else {
 		r2 = ret.Error(2)
 	}
@@ -125,13 +126,14 @@ type MockCipher_GetEncryptor_Call struct {
 
 // GetEncryptor is a helper method to define mock.On call
 //   - ezID int64
-func (_e *MockCipher_Expecter) GetEncryptor(ezID interface{}) *MockCipher_GetEncryptor_Call {
+//   - collectionID int64
-	return &MockCipher_GetEncryptor_Call{Call: _e.mock.On("GetEncryptor", ezID)}
+func (_e *MockCipher_Expecter) GetEncryptor(ezID interface{}, collectionID interface{}) *MockCipher_GetEncryptor_Call {
+	return &MockCipher_GetEncryptor_Call{Call: _e.mock.On("GetEncryptor", ezID, collectionID)}
 }
 
-func (_c *MockCipher_GetEncryptor_Call) Run(run func(ezID int64)) *MockCipher_GetEncryptor_Call {
+func (_c *MockCipher_GetEncryptor_Call) Run(run func(ezID int64, collectionID int64)) *MockCipher_GetEncryptor_Call {
 	_c.Call.Run(func(args mock.Arguments) {
-		run(args[0].(int64))
+		run(args[0].(int64), args[1].(int64))
 	})
 	return _c
 }
@@ -141,22 +143,22 @@ func (_c *MockCipher_GetEncryptor_Call) Return(encryptor hook.Encryptor, safeKey
 	return _c
 }
 
-func (_c *MockCipher_GetEncryptor_Call) RunAndReturn(run func(int64) (hook.Encryptor, []byte, error)) *MockCipher_GetEncryptor_Call {
+func (_c *MockCipher_GetEncryptor_Call) RunAndReturn(run func(int64, int64) (hook.Encryptor, []byte, error)) *MockCipher_GetEncryptor_Call {
 	_c.Call.Return(run)
 	return _c
 }
 
-// GetUnsafeKey provides a mock function with given fields: ezID
+// GetUnsafeKey provides a mock function with given fields: ezID, collectionID
-func (_m *MockCipher) GetUnsafeKey(ezID int64) []byte {
+func (_m *MockCipher) GetUnsafeKey(ezID int64, collectionID int64) []byte {
-	ret := _m.Called(ezID)
+	ret := _m.Called(ezID, collectionID)
 
 	if len(ret) == 0 {
 		panic("no return value specified for GetUnsafeKey")
 	}
 
 	var r0 []byte
-	if rf, ok := ret.Get(0).(func(int64) []byte); ok {
+	if rf, ok := ret.Get(0).(func(int64, int64) []byte); ok {
-		r0 = rf(ezID)
+		r0 = rf(ezID, collectionID)
 	} else {
 		if ret.Get(0) != nil {
 			r0 = ret.Get(0).([]byte)
@@ -173,13 +175,14 @@ type MockCipher_GetUnsafeKey_Call struct {
 
 // GetUnsafeKey is a helper method to define mock.On call
 //   - ezID int64
-func (_e *MockCipher_Expecter) GetUnsafeKey(ezID interface{}) *MockCipher_GetUnsafeKey_Call {
+//   - collectionID int64
-	return &MockCipher_GetUnsafeKey_Call{Call: _e.mock.On("GetUnsafeKey", ezID)}
+func (_e *MockCipher_Expecter) GetUnsafeKey(ezID interface{}, collectionID interface{}) *MockCipher_GetUnsafeKey_Call {
+	return &MockCipher_GetUnsafeKey_Call{Call: _e.mock.On("GetUnsafeKey", ezID, collectionID)}
 }
 
-func (_c *MockCipher_GetUnsafeKey_Call) Run(run func(ezID int64)) *MockCipher_GetUnsafeKey_Call {
+func (_c *MockCipher_GetUnsafeKey_Call) Run(run func(ezID int64, collectionID int64)) *MockCipher_GetUnsafeKey_Call {
 	_c.Call.Run(func(args mock.Arguments) {
-		run(args[0].(int64))
+		run(args[0].(int64), args[1].(int64))
 	})
 	return _c
 }
@@ -189,7 +192,7 @@ func (_c *MockCipher_GetUnsafeKey_Call) Return(_a0 []byte) *MockCipher_GetUnsafe
 	return _c
 }
 
-func (_c *MockCipher_GetUnsafeKey_Call) RunAndReturn(run func(int64) []byte) *MockCipher_GetUnsafeKey_Call {
+func (_c *MockCipher_GetUnsafeKey_Call) RunAndReturn(run func(int64, int64) []byte) *MockCipher_GetUnsafeKey_Call {
 	_c.Call.Return(run)
 	return _c
 }

pkg/proto/messages.proto

@@ -297,6 +297,7 @@ message ResourceKey {
 // CipherHeader is the header of a message that is encrypted.
 message CipherHeader {
     int64 ez_id = 1; // related to the encryption zone id
-    bytes safe_key = 2; // the safe key
+    int64 collection_id = 2; // related to the collection id
-    int64 payload_bytes = 3; // the size of the payload before encryption
+    bytes safe_key = 3; // the safe key
+    int64 payload_bytes = 4; // the size of the payload before encryption
 }

pkg/proto/messagespb/messages.pb.go

@@ -2041,8 +2041,9 @@ type CipherHeader struct {
 	unknownFields protoimpl.UnknownFields
 
 	EzId         int64  `protobuf:"varint,1,opt,name=ez_id,json=ezId,proto3" json:"ez_id,omitempty"`                         // related to the encryption zone id
-	SafeKey      []byte `protobuf:"bytes,2,opt,name=safe_key,json=safeKey,proto3" json:"safe_key,omitempty"`                 // the safe key
+	CollectionId int64  `protobuf:"varint,2,opt,name=collection_id,json=collectionId,proto3" json:"collection_id,omitempty"` // related to the collection id
-	PayloadBytes int64  `protobuf:"varint,3,opt,name=payload_bytes,json=payloadBytes,proto3" json:"payload_bytes,omitempty"` // the size of the payload before encryption
+	SafeKey      []byte `protobuf:"bytes,3,opt,name=safe_key,json=safeKey,proto3" json:"safe_key,omitempty"`                 // the safe key
+	PayloadBytes int64  `protobuf:"varint,4,opt,name=payload_bytes,json=payloadBytes,proto3" json:"payload_bytes,omitempty"` // the size of the payload before encryption
 }
 
 func (x *CipherHeader) Reset() {
@@ -2084,6 +2085,13 @@ func (x *CipherHeader) GetEzId() int64 {
 	return 0
 }
 
+func (x *CipherHeader) GetCollectionId() int64 {
+	if x != nil {
+		return x.CollectionId
+	}
+	return 0
+}
+
 func (x *CipherHeader) GetSafeKey() []byte {
 	if x != nil {
 		return x.SafeKey
@@ -2294,50 +2302,53 @@ var file_messages_proto_rawDesc = []byte{
 	0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, 0x52,
 	0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x44, 0x6f, 0x6d, 0x61, 0x69, 0x6e, 0x52, 0x06, 0x64,
 	0x6f, 0x6d, 0x61, 0x69, 0x6e, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x02, 0x20, 0x01,
-	0x28, 0x09, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x22, 0x63, 0x0a, 0x0c, 0x43, 0x69, 0x70, 0x68, 0x65,
+	0x28, 0x09, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x22, 0x88, 0x01, 0x0a, 0x0c, 0x43, 0x69, 0x70, 0x68,
-	0x72, 0x48, 0x65, 0x61, 0x64, 0x65, 0x72, 0x12, 0x13, 0x0a, 0x05, 0x65, 0x7a, 0x5f, 0x69, 0x64,
+	0x65, 0x72, 0x48, 0x65, 0x61, 0x64, 0x65, 0x72, 0x12, 0x13, 0x0a, 0x05, 0x65, 0x7a, 0x5f, 0x69,
-	0x18, 0x01, 0x20, 0x01, 0x28, 0x03, 0x52, 0x04, 0x65, 0x7a, 0x49, 0x64, 0x12, 0x19, 0x0a, 0x08,
+	0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x03, 0x52, 0x04, 0x65, 0x7a, 0x49, 0x64, 0x12, 0x23, 0x0a,
-	0x73, 0x61, 0x66, 0x65, 0x5f, 0x6b, 0x65, 0x79, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x07,
+	0x0d, 0x63, 0x6f, 0x6c, 0x6c, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x69, 0x64, 0x18, 0x02,
-	0x73, 0x61, 0x66, 0x65, 0x4b, 0x65, 0x79, 0x12, 0x23, 0x0a, 0x0d, 0x70, 0x61, 0x79, 0x6c, 0x6f,
+	0x20, 0x01, 0x28, 0x03, 0x52, 0x0c, 0x63, 0x6f, 0x6c, 0x6c, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e,
-	0x61, 0x64, 0x5f, 0x62, 0x79, 0x74, 0x65, 0x73, 0x18, 0x03, 0x20, 0x01, 0x28, 0x03, 0x52, 0x0c,
+	0x49, 0x64, 0x12, 0x19, 0x0a, 0x08, 0x73, 0x61, 0x66, 0x65, 0x5f, 0x6b, 0x65, 0x79, 0x18, 0x03,
-	0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x42, 0x79, 0x74, 0x65, 0x73, 0x2a, 0x9a, 0x02, 0x0a,
+	0x20, 0x01, 0x28, 0x0c, 0x52, 0x07, 0x73, 0x61, 0x66, 0x65, 0x4b, 0x65, 0x79, 0x12, 0x23, 0x0a,
-	0x0b, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x54, 0x79, 0x70, 0x65, 0x12, 0x0b, 0x0a, 0x07,
+	0x0d, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x5f, 0x62, 0x79, 0x74, 0x65, 0x73, 0x18, 0x04,
-	0x55, 0x6e, 0x6b, 0x6e, 0x6f, 0x77, 0x6e, 0x10, 0x00, 0x12, 0x0c, 0x0a, 0x08, 0x54, 0x69, 0x6d,
+	0x20, 0x01, 0x28, 0x03, 0x52, 0x0c, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x42, 0x79, 0x74,
-	0x65, 0x54, 0x69, 0x63, 0x6b, 0x10, 0x01, 0x12, 0x0a, 0x0a, 0x06, 0x49, 0x6e, 0x73, 0x65, 0x72,
+	0x65, 0x73, 0x2a, 0x9a, 0x02, 0x0a, 0x0b, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x54, 0x79,
-	0x74, 0x10, 0x02, 0x12, 0x0a, 0x0a, 0x06, 0x44, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x10, 0x03, 0x12,
+	0x70, 0x65, 0x12, 0x0b, 0x0a, 0x07, 0x55, 0x6e, 0x6b, 0x6e, 0x6f, 0x77, 0x6e, 0x10, 0x00, 0x12,
-	0x09, 0x0a, 0x05, 0x46, 0x6c, 0x75, 0x73, 0x68, 0x10, 0x04, 0x12, 0x14, 0x0a, 0x10, 0x43, 0x72,
+	0x0c, 0x0a, 0x08, 0x54, 0x69, 0x6d, 0x65, 0x54, 0x69, 0x63, 0x6b, 0x10, 0x01, 0x12, 0x0a, 0x0a,
-	0x65, 0x61, 0x74, 0x65, 0x43, 0x6f, 0x6c, 0x6c, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x10, 0x05,
+	0x06, 0x49, 0x6e, 0x73, 0x65, 0x72, 0x74, 0x10, 0x02, 0x12, 0x0a, 0x0a, 0x06, 0x44, 0x65, 0x6c,
-	0x12, 0x12, 0x0a, 0x0e, 0x44, 0x72, 0x6f, 0x70, 0x43, 0x6f, 0x6c, 0x6c, 0x65, 0x63, 0x74, 0x69,
+	0x65, 0x74, 0x65, 0x10, 0x03, 0x12, 0x09, 0x0a, 0x05, 0x46, 0x6c, 0x75, 0x73, 0x68, 0x10, 0x04,
-	0x6f, 0x6e, 0x10, 0x06, 0x12, 0x13, 0x0a, 0x0f, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65, 0x50, 0x61,
+	0x12, 0x14, 0x0a, 0x10, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65, 0x43, 0x6f, 0x6c, 0x6c, 0x65, 0x63,
-	0x72, 0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x10, 0x07, 0x12, 0x11, 0x0a, 0x0d, 0x44, 0x72, 0x6f,
+	0x74, 0x69, 0x6f, 0x6e, 0x10, 0x05, 0x12, 0x12, 0x0a, 0x0e, 0x44, 0x72, 0x6f, 0x70, 0x43, 0x6f,
-	0x70, 0x50, 0x61, 0x72, 0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x10, 0x08, 0x12, 0x0f, 0x0a, 0x0b,
+	0x6c, 0x6c, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x10, 0x06, 0x12, 0x13, 0x0a, 0x0f, 0x43, 0x72,
-	0x4d, 0x61, 0x6e, 0x75, 0x61, 0x6c, 0x46, 0x6c, 0x75, 0x73, 0x68, 0x10, 0x09, 0x12, 0x11, 0x0a,
+	0x65, 0x61, 0x74, 0x65, 0x50, 0x61, 0x72, 0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x10, 0x07, 0x12,
-	0x0d, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65, 0x53, 0x65, 0x67, 0x6d, 0x65, 0x6e, 0x74, 0x10, 0x0a,
+	0x11, 0x0a, 0x0d, 0x44, 0x72, 0x6f, 0x70, 0x50, 0x61, 0x72, 0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e,
-	0x12, 0x0a, 0x0a, 0x06, 0x49, 0x6d, 0x70, 0x6f, 0x72, 0x74, 0x10, 0x0b, 0x12, 0x10, 0x0a, 0x0c,
+	0x10, 0x08, 0x12, 0x0f, 0x0a, 0x0b, 0x4d, 0x61, 0x6e, 0x75, 0x61, 0x6c, 0x46, 0x6c, 0x75, 0x73,
-	0x53, 0x63, 0x68, 0x65, 0x6d, 0x61, 0x43, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x10, 0x0c, 0x12, 0x0d,
+	0x68, 0x10, 0x09, 0x12, 0x11, 0x0a, 0x0d, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65, 0x53, 0x65, 0x67,
-	0x0a, 0x08, 0x42, 0x65, 0x67, 0x69, 0x6e, 0x54, 0x78, 0x6e, 0x10, 0x84, 0x07, 0x12, 0x0e, 0x0a,
+	0x6d, 0x65, 0x6e, 0x74, 0x10, 0x0a, 0x12, 0x0a, 0x0a, 0x06, 0x49, 0x6d, 0x70, 0x6f, 0x72, 0x74,
-	0x09, 0x43, 0x6f, 0x6d, 0x6d, 0x69, 0x74, 0x54, 0x78, 0x6e, 0x10, 0x85, 0x07, 0x12, 0x10, 0x0a,
+	0x10, 0x0b, 0x12, 0x10, 0x0a, 0x0c, 0x53, 0x63, 0x68, 0x65, 0x6d, 0x61, 0x43, 0x68, 0x61, 0x6e,
-	0x0b, 0x52, 0x6f, 0x6c, 0x6c, 0x62, 0x61, 0x63, 0x6b, 0x54, 0x78, 0x6e, 0x10, 0x86, 0x07, 0x12,
+	0x67, 0x65, 0x10, 0x0c, 0x12, 0x0d, 0x0a, 0x08, 0x42, 0x65, 0x67, 0x69, 0x6e, 0x54, 0x78, 0x6e,
-	0x08, 0x0a, 0x03, 0x54, 0x78, 0x6e, 0x10, 0xe7, 0x07, 0x2a, 0x82, 0x01, 0x0a, 0x08, 0x54, 0x78,
+	0x10, 0x84, 0x07, 0x12, 0x0e, 0x0a, 0x09, 0x43, 0x6f, 0x6d, 0x6d, 0x69, 0x74, 0x54, 0x78, 0x6e,
-	0x6e, 0x53, 0x74, 0x61, 0x74, 0x65, 0x12, 0x0e, 0x0a, 0x0a, 0x54, 0x78, 0x6e, 0x55, 0x6e, 0x6b,
+	0x10, 0x85, 0x07, 0x12, 0x10, 0x0a, 0x0b, 0x52, 0x6f, 0x6c, 0x6c, 0x62, 0x61, 0x63, 0x6b, 0x54,
-	0x6e, 0x6f, 0x77, 0x6e, 0x10, 0x00, 0x12, 0x0c, 0x0a, 0x08, 0x54, 0x78, 0x6e, 0x42, 0x65, 0x67,
+	0x78, 0x6e, 0x10, 0x86, 0x07, 0x12, 0x08, 0x0a, 0x03, 0x54, 0x78, 0x6e, 0x10, 0xe7, 0x07, 0x2a,
-	0x69, 0x6e, 0x10, 0x01, 0x12, 0x0f, 0x0a, 0x0b, 0x54, 0x78, 0x6e, 0x49, 0x6e, 0x46, 0x6c, 0x69,
+	0x82, 0x01, 0x0a, 0x08, 0x54, 0x78, 0x6e, 0x53, 0x74, 0x61, 0x74, 0x65, 0x12, 0x0e, 0x0a, 0x0a,
-	0x67, 0x68, 0x74, 0x10, 0x02, 0x12, 0x0f, 0x0a, 0x0b, 0x54, 0x78, 0x6e, 0x4f, 0x6e, 0x43, 0x6f,
+	0x54, 0x78, 0x6e, 0x55, 0x6e, 0x6b, 0x6e, 0x6f, 0x77, 0x6e, 0x10, 0x00, 0x12, 0x0c, 0x0a, 0x08,
-	0x6d, 0x6d, 0x69, 0x74, 0x10, 0x03, 0x12, 0x10, 0x0a, 0x0c, 0x54, 0x78, 0x6e, 0x43, 0x6f, 0x6d,
+	0x54, 0x78, 0x6e, 0x42, 0x65, 0x67, 0x69, 0x6e, 0x10, 0x01, 0x12, 0x0f, 0x0a, 0x0b, 0x54, 0x78,
-	0x6d, 0x69, 0x74, 0x74, 0x65, 0x64, 0x10, 0x04, 0x12, 0x11, 0x0a, 0x0d, 0x54, 0x78, 0x6e, 0x4f,
+	0x6e, 0x49, 0x6e, 0x46, 0x6c, 0x69, 0x67, 0x68, 0x74, 0x10, 0x02, 0x12, 0x0f, 0x0a, 0x0b, 0x54,
-	0x6e, 0x52, 0x6f, 0x6c, 0x6c, 0x62, 0x61, 0x63, 0x6b, 0x10, 0x05, 0x12, 0x11, 0x0a, 0x0d, 0x54,
+	0x78, 0x6e, 0x4f, 0x6e, 0x43, 0x6f, 0x6d, 0x6d, 0x69, 0x74, 0x10, 0x03, 0x12, 0x10, 0x0a, 0x0c,
-	0x78, 0x6e, 0x52, 0x6f, 0x6c, 0x6c, 0x62, 0x61, 0x63, 0x6b, 0x65, 0x64, 0x10, 0x06, 0x2a, 0x6c,
+	0x54, 0x78, 0x6e, 0x43, 0x6f, 0x6d, 0x6d, 0x69, 0x74, 0x74, 0x65, 0x64, 0x10, 0x04, 0x12, 0x11,
-	0x0a, 0x0e, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x44, 0x6f, 0x6d, 0x61, 0x69, 0x6e,
+	0x0a, 0x0d, 0x54, 0x78, 0x6e, 0x4f, 0x6e, 0x52, 0x6f, 0x6c, 0x6c, 0x62, 0x61, 0x63, 0x6b, 0x10,
-	0x12, 0x19, 0x0a, 0x15, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x44, 0x6f, 0x6d, 0x61,
+	0x05, 0x12, 0x11, 0x0a, 0x0d, 0x54, 0x78, 0x6e, 0x52, 0x6f, 0x6c, 0x6c, 0x62, 0x61, 0x63, 0x6b,
-	0x69, 0x6e, 0x55, 0x6e, 0x6b, 0x6e, 0x6f, 0x77, 0x6e, 0x10, 0x00, 0x12, 0x1d, 0x0a, 0x19, 0x52,
+	0x65, 0x64, 0x10, 0x06, 0x2a, 0x6c, 0x0a, 0x0e, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65,
-	0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x44, 0x6f, 0x6d, 0x61, 0x69, 0x6e, 0x49, 0x6d, 0x70,
+	0x44, 0x6f, 0x6d, 0x61, 0x69, 0x6e, 0x12, 0x19, 0x0a, 0x15, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72,
-	0x6f, 0x72, 0x74, 0x4a, 0x6f, 0x62, 0x49, 0x44, 0x10, 0x01, 0x12, 0x20, 0x0a, 0x1c, 0x52, 0x65,
+	0x63, 0x65, 0x44, 0x6f, 0x6d, 0x61, 0x69, 0x6e, 0x55, 0x6e, 0x6b, 0x6e, 0x6f, 0x77, 0x6e, 0x10,
-	0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x44, 0x6f, 0x6d, 0x61, 0x69, 0x6e, 0x43, 0x6f, 0x6c, 0x6c,
+	0x00, 0x12, 0x1d, 0x0a, 0x19, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x44, 0x6f, 0x6d,
-	0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x4e, 0x61, 0x6d, 0x65, 0x10, 0x02, 0x42, 0x35, 0x5a, 0x33,
+	0x61, 0x69, 0x6e, 0x49, 0x6d, 0x70, 0x6f, 0x72, 0x74, 0x4a, 0x6f, 0x62, 0x49, 0x44, 0x10, 0x01,
-	0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x6d, 0x69, 0x6c, 0x76, 0x75,
+	0x12, 0x20, 0x0a, 0x1c, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x44, 0x6f, 0x6d, 0x61,
-	0x73, 0x2d, 0x69, 0x6f, 0x2f, 0x6d, 0x69, 0x6c, 0x76, 0x75, 0x73, 0x2f, 0x70, 0x6b, 0x67, 0x2f,
+	0x69, 0x6e, 0x43, 0x6f, 0x6c, 0x6c, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x4e, 0x61, 0x6d, 0x65,
-	0x76, 0x32, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65,
+	0x10, 0x02, 0x42, 0x35, 0x5a, 0x33, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d,
-	0x73, 0x70, 0x62, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+	0x2f, 0x6d, 0x69, 0x6c, 0x76, 0x75, 0x73, 0x2d, 0x69, 0x6f, 0x2f, 0x6d, 0x69, 0x6c, 0x76, 0x75,
+	0x73, 0x2f, 0x70, 0x6b, 0x67, 0x2f, 0x76, 0x32, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2f, 0x6d,
+	0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x70, 0x62, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f,
+	0x33,
 }
 
 var (

pkg/streaming/util/message/builder.go

@@ -262,7 +262,7 @@ func (b *mutableMesasgeBuilder[H, B]) build() (*messageImpl, error) {
 		}
 
 		cipher := mustGetCipher()
-		encryptor, safeKey, err := cipher.GetEncryptor(b.cipherConfig.EzID)
+		encryptor, safeKey, err := cipher.GetEncryptor(b.cipherConfig.EzID, b.cipherConfig.CollectionID)
 		if err != nil {
 			return nil, errors.Wrap(err, "failed to get encryptor")
 		}
@@ -272,6 +272,7 @@ func (b *mutableMesasgeBuilder[H, B]) build() (*messageImpl, error) {
 		}
 		ch, err := EncodeProto(&messagespb.CipherHeader{
 			EzId:         b.cipherConfig.EzID,
+			CollectionId: b.cipherConfig.CollectionID,
 			SafeKey:      safeKey,
 			PayloadBytes: int64(payloadBytes),
 		})

pkg/streaming/util/message/cipher.go

@@ -31,4 +31,7 @@ func mustGetCipher() hook.Cipher {
 type CipherConfig struct {
 	// EzID is the encryption zone ID.
 	EzID int64
+
+	// Collection ID
+	CollectionID int64
 }

pkg/streaming/util/message/message_impl.go

@@ -33,7 +33,7 @@ func (m *messageImpl) Version() Version {
 func (m *messageImpl) Payload() []byte {
 	if ch := m.cipherHeader(); ch != nil {
 		cipher := mustGetCipher()
-		decryptor, err := cipher.GetDecryptor(ch.EzId, ch.SafeKey)
+		decryptor, err := cipher.GetDecryptor(ch.EzId, ch.CollectionId, ch.SafeKey)
 		if err != nil {
 			panic(fmt.Sprintf("can not get decryptor for message: %s", err))
 		}

pkg/streaming/util/message/message_test.go

@@ -99,8 +99,8 @@ func TestCiper(t *testing.T) {
 	d.EXPECT().Decrypt(mock.Anything).RunAndReturn(func(b []byte) ([]byte, error) {
 		return b[3:], nil
 	})
-	c.EXPECT().GetEncryptor(mock.Anything).Return(e, []byte("123"), nil)
+	c.EXPECT().GetEncryptor(mock.Anything, mock.Anything).Return(e, []byte("123"), nil)
-	c.EXPECT().GetDecryptor(mock.Anything, mock.Anything).Return(d, nil)
+	c.EXPECT().GetDecryptor(mock.Anything, mock.Anything, mock.Anything).Return(d, nil)
 	RegisterCipher(c)
 
 	msg, _ := builder.WithCipher(&CipherConfig{