Persist GroupKeys

package.json

@@ -151,7 +151,9 @@
     "@ethersproject/wallet": "^5.1.0",
     "@ethersproject/web": "^5.1.0",
     "debug": "^4.3.2",
+    "env-paths": "^2.2.1",
     "eventemitter3": "^4.0.7",
+    "idb-keyval": "^5.0.5",
     "lodash": "^4.17.21",
     "mem": "^8.1.1",
     "node-abort-controller": "^1.2.1",
@@ -165,6 +167,8 @@
     "qs": "^6.10.1",
     "quick-lru": "^6.0.0",
     "readable-stream": "^3.6.0",
+    "sqlite": "^4.0.21",
+    "sqlite3": "^5.0.2",
     "streamr-client-protocol": "^8.0.2",
     "streamr-test-utils": "^1.3.2",
     "ts-toolbelt": "^9.6.0",

src/Connection.ts

@@ -425,7 +425,7 @@ export default class Connection extends EventEmitter {
         }
     }
 
-    enableAutoDisconnect(autoDisconnect = true) {
+    enableAutoDisconnect(autoDisconnect: boolean | number = true) {
         let delay
         if (typeof autoDisconnect === 'number') {
             delay = autoDisconnect
@@ -443,7 +443,7 @@ export default class Connection extends EventEmitter {
         }
     }
 
-    enableAutoConnect(autoConnect = true) {
+    enableAutoConnect(autoConnect: boolean | number = true) {
         autoConnect = !!autoConnect // eslint-disable-line no-param-reassign
         if (this.options.autoConnect && !autoConnect) {
             this.didDisableAutoConnect = true

src/StreamrClient.ts

@@ -22,8 +22,7 @@ import { DataUnion, DataUnionDeployOptions } from './dataunion/DataUnion'
 import { BigNumber } from '@ethersproject/bignumber'
 import { getAddress } from '@ethersproject/address'
 import { Contract } from '@ethersproject/contracts'
-import { StreamPartDefinition } from './stream'
-import type { GroupKey } from './stream/Encryption'
+import { StreamPartDefinition, GroupKey } from './stream'
 
 // TODO get metadata type from streamr-protocol-js project (it doesn't export the type definitions yet)
 export type OnMessageCallback = MaybeAsync<(message: any, metadata: any) => void>

src/index.ts

@@ -10,7 +10,7 @@ import { StreamrClient } from './StreamrClient'
 export * from './StreamrClient'
 export * from './Config'
 export * from './stream'
-export * from './stream/Encryption'
+export * from './stream/encryption/Encryption'
 export * from './stream/StreamPart'
 export * from './stream/StorageNode'
 export * from './subscribe'

src/publish/Encrypt.ts

@@ -1,20 +1,18 @@
 import { MessageLayer } from 'streamr-client-protocol'
 
-import EncryptionUtil from '../stream/Encryption'
+import EncryptionUtil from '../stream/encryption/Encryption'
 import { Stream } from '../stream'
 import { StreamrClient } from '../StreamrClient'
-import { PublisherKeyExhange } from '../stream/KeyExchange'
+import { PublisherKeyExhange } from '../stream/encryption/KeyExchangePublisher'
 
 const { StreamMessage } = MessageLayer
 
-type PublisherKeyExhangeAPI = ReturnType<typeof PublisherKeyExhange>
-
 export default function Encrypt(client: StreamrClient) {
-    let publisherKeyExchange: ReturnType<typeof PublisherKeyExhange>
+    let publisherKeyExchange: PublisherKeyExhange
 
     function getPublisherKeyExchange() {
         if (!publisherKeyExchange) {
-            publisherKeyExchange = PublisherKeyExhange(client, {
+            publisherKeyExchange = new PublisherKeyExhange(client, {
                 groupKeys: {
                     ...client.options.groupKeys,
                 }
@@ -28,9 +26,19 @@ export default function Encrypt(client: StreamrClient) {
             return
         }
 
+        const { messageType } = streamMessage
+        if (
+            messageType === StreamMessage.MESSAGE_TYPES.GROUP_KEY_RESPONSE
+            || messageType === StreamMessage.MESSAGE_TYPES.GROUP_KEY_REQUEST
+            || messageType === StreamMessage.MESSAGE_TYPES.GROUP_KEY_ERROR_RESPONSE
+        ) {
+            // never encrypt
+            return
+        }
+
         if (
             !stream.requireEncryptedData
-            && !getPublisherKeyExchange().hasAnyGroupKey(stream.id)
+            && !(await (getPublisherKeyExchange().hasAnyGroupKey(stream.id)))
         ) {
             // not needed
             return
@@ -41,17 +49,21 @@ export default function Encrypt(client: StreamrClient) {
         }
 
         const [groupKey, nextGroupKey] = await getPublisherKeyExchange().useGroupKey(stream.id)
+        if (!groupKey) {
+            throw new Error(`Tried to use group key but no group key found for stream: ${stream.id}`)
+        }
+
         await EncryptionUtil.encryptStreamMessage(streamMessage, groupKey, nextGroupKey)
     }
 
     return Object.assign(encrypt, {
-        setNextGroupKey(...args: Parameters<PublisherKeyExhangeAPI['setNextGroupKey']>) {
+        setNextGroupKey(...args: Parameters<PublisherKeyExhange['setNextGroupKey']>) {
             return getPublisherKeyExchange().setNextGroupKey(...args)
         },
-        rotateGroupKey(...args: Parameters<PublisherKeyExhangeAPI['rotateGroupKey']>) {
+        rotateGroupKey(...args: Parameters<PublisherKeyExhange['rotateGroupKey']>) {
             return getPublisherKeyExchange().rotateGroupKey(...args)
         },
-        rekey(...args: Parameters<PublisherKeyExhangeAPI['rekey']>) {
+        rekey(...args: Parameters<PublisherKeyExhange['rekey']>) {
             return getPublisherKeyExchange().rekey(...args)
         },
         start() {

src/rest/StreamEndpoints.ts

@@ -8,7 +8,7 @@ import { getEndpointUrl } from '../utils'
 import { validateOptions } from '../stream/utils'
 import { Stream, StreamOperation, StreamProperties } from '../stream'
 import { StreamPart } from '../stream/StreamPart'
-import { isKeyExchangeStream } from '../stream/KeyExchange'
+import { isKeyExchangeStream } from '../stream/encryption/KeyExchangeUtils'
 
 import authFetch, { ErrorCode, NotFoundError } from './authFetch'
 import { EthereumAddress } from '../types'