Postgres POC (duplicated SQLite code)

* use aeson fork with the option to encode/decode nullary constructors as empty objects

* update dependencies

.github/changelog_conf.json

@@ -1,4 +1,4 @@
 {
   "template": "${{UNCATEGORIZED}}",
-  "pr_template": "- ${{TITLE}}\n"
+  "pr_template": "- ${{TITLE}}"
 }

.github/workflows/build.yml

@@ -19,7 +19,7 @@ jobs:
       - name: Setup Stack
         uses: haskell/actions/setup@v1
         with:
-          ghc-version: '8.8.4'
+          ghc-version: '8.10.7'
           enable-stack: true
           stack-version: 'latest'
 
@@ -31,11 +31,12 @@ jobs:
 
       - name: Build & test
         id: build_test
+        shell: bash
         run: |
           stack build --test --force-dirty
           install_root=$(stack path --local-install-root)
           mv ${install_root}/bin/smp-server smp-server-ubuntu-20_04-x86-64
-      
+
       - name: Build changelog
         if: startsWith(github.ref, 'refs/tags/v')
         id: build_changelog
@@ -47,15 +48,28 @@ jobs:
           commitMode: true
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-      
+
+      - name: Extract release candidate
+        if: startsWith(github.ref, 'refs/tags/v')
+        id: extract_release_candidate
+        shell: bash
+        run: |
+          if [[ ${GITHUB_REF} == *rc* ]]; then
+            echo "::set-output name=release_candidate::true"
+          else
+            echo "::set-output name=release_candidate::false"
+          fi
+
       - name: Create release
         if: startsWith(github.ref, 'refs/tags/v')
         uses: softprops/action-gh-release@v1
         with:
           body: |
             See full changelog [here](https://github.com/simplex-chat/simplexmq/blob/master/CHANGELOG.md).
-            Commits, chronological:
+
+            Commits:
             ${{ steps.build_changelog.outputs.changelog }}
+          prerelease: ${{ steps.extract_release_candidate.outputs.release_candidate }}
           files: |
             LICENSE
             smp-server-ubuntu-20_04-x86-64

.gitignore

@@ -3,3 +3,4 @@
 *.db.bak
 *.session.sql
 tests/tmp
+dist-newstyle/

CHANGELOG.md

@@ -1,3 +1,16 @@
+# 1.0.2
+
+General:
+- Enable TLS 1.3 parameters for TLS handshake (server and client).
+- Switch from hs-tls fork to original repo now that it supports getFinished and getPeerFinished APIs for both TLS 1.2 and TLS 1.3.
+
+SMP server:
+- Perform TLS handshake in a separate thread per-connection.
+
+SMP agent:
+- Cease attempts to send HELLO after one week timeout.
+- Coalesce requests to connect to SMP servers, to have 1 connection per server.
+
 # 1.0.1
 
 SMP server:

README.md

@@ -35,6 +35,8 @@ SMP server uses in-memory persistence with an optional append-only log of create
 
 To enable store log, initialize server using `smp-server -l` command, or modify `smp-server.ini` created during initialization (uncomment `enable: on` option in the store log section). Use `smp-server --help` for other usage tips.
 
+> **Please note:** On initialization SMP server creates a chain of two certificates: a self-signed CA certificate ("offline") and a server certificate used for TLS handshake ("online"). **You should store CA certificate private key securely and delete it from the server. If server TLS credential is compromised this key can be used to sign a new one, keeping the same server identity and established connections.** CA private key location by default is `/etc/opt/simplex/ca.key`.
+
 SMP server implements [SMP protocol](https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md).
 
 #### Running SMP server on MacOS
@@ -96,10 +98,12 @@ Deployment on Linode is performed via StackScripts, which serve as recipes for L
 - Create a Linode account or login with an already existing one.
 - Open [SMP server StackScript](https://cloud.linode.com/stackscripts/748014) and click "Deploy New Linode".
 - You can optionally configure the following parameters:
-    - [SMP Server store log](#SMP-server) flag for queue persistence on server restart (recommended).
-    - [Linode API token](https://www.linode.com/docs/guides/getting-started-with-the-linode-api#get-an-access-token) for attaching server info as tags to Linode (server address, fingerprint, version) and adding A record to your 2nd level domain (Note: 2nd level e.g. `example.com` domain should be [created](https://cloud.linode.com/domains/create) in your account prior to deployment). The API token access scope should be read/write access to "linodes" (to create tags), and "domains" (to add A record for the 3rd level domain, e.g. `smp`).
-    - Domain name to use instead of Linode ip address, e.g. `smp.example.com`.
-- Choose the region and plan according to your requirements (for regular use Shared CPU Nanode should be sufficient).
+    - SMP Server store log flag for queue persistence on server restart, recommended.
+    - [Linode API token](https://www.linode.com/docs/guides/getting-started-with-the-linode-api#get-an-access-token) to attach server address etc. as tags to Linode and to add A record to your 2nd level domain (e.g. `example.com` [domain should be created](https://cloud.linode.com/domains/create) in your account prior to deployment). The API token access scopes:
+      - read/write for "linodes"
+      - read/write for "domains"
+    - Domain name to use instead of Linode IP address, e.g. `smp1.example.com`.
+- Choose the region and plan, Shared CPU Nanode with 1Gb is sufficient.
 - Provide ssh key to be able to connect to your Linode via ssh. If you haven't provided a Linode API token this step is required to login to your Linode and get the server's fingerprint either from the welcome message or from the file `/etc/opt/simplex/fingerprint` after server starts. See [Linode's guide on ssh](https://www.linode.com/docs/guides/use-public-key-authentication-with-ssh/) .
 - Deploy your Linode. After it starts wait for SMP server to start and for tags to appear (if a Linode API token was provided). It may take up to 5 minutes depending on the connection speed on the Linode. Connecting Linode IP address to provided domain name may take some additional time.
 - Get `address` and `fingerprint` either from Linode tags (click on a tag and copy it's value from the browser search panel) or via ssh.
@@ -121,11 +125,18 @@ To deploy SMP server use [SimpleX Server 1-click app](https://marketplace.digita
 - Click 'Create SimpleX server Droplet' button.
 - Choose the region and plan according to your requirements (Basic plan should be sufficient).
 - Finalize Droplet creation.
-- Open "Console" on your Droplet management page to get SMP server fingerprint - either from the welcome message or from `/etc/opt/simplex/fingerprint`. Alternatively you can manually SSH to created Droplet, see [instruction](https://docs.digitalocean.com/products/droplets/how-to/connect-with-ssh/).
+- Open "Console" on your Droplet management page to get SMP server fingerprint - either from the welcome message or from `/etc/opt/simplex/fingerprint`. Alternatively you can manually SSH to created Droplet, see [DigitalOcean instruction](https://docs.digitalocean.com/products/droplets/how-to/connect-with-ssh/).
 - Great, your own SMP server is ready! Use `smp://<fingerprint>@<ip_address>` as SMP server address in the client.
 
 Please submit an [issue](https://github.com/simplex-chat/simplexmq/issues) if any problems occur.
 
+> **Please note:** SMP server uses server address as a Common Name for server certificate generated during initialization. If you would like your server address to be FQDN instead of IP address, you can log in to your Droplet and run the commands below to re-initialize the server. Alternatively you can use [Linode StackScript](https://cloud.linode.com/stackscripts/748014) which allows this parameterization.
+
+```sh
+smp-server delete
+smp-server init [-l] -n <fqdn>
+```
+
 ## SMP server design
 
 ![SMP server design](https://raw.githubusercontent.com/simplex-chat/simplexmq/master/design/server.svg)

apps/smp-agent/Main.hs

@@ -6,8 +6,8 @@ module Main where
 
 import Control.Logger.Simple
 import qualified Data.List.NonEmpty as L
-import Simplex.Messaging.Agent (runSMPAgent)
 import Simplex.Messaging.Agent.Env.SQLite
+import Simplex.Messaging.Agent.Server (runSMPAgent)
 import Simplex.Messaging.Transport (TLS, Transport (..))
 
 cfg :: AgentConfig

apps/smp-server/Main.hs

@@ -22,12 +22,13 @@ import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Server (runSMPServer)
 import Simplex.Messaging.Server.Env.STM
 import Simplex.Messaging.Server.StoreLog (StoreLog, openReadStoreLog, storeLogFilePath)
-import Simplex.Messaging.Transport (ATransport (..), TLS, Transport (..), loadFingerprint, simplexMQVersion)
+import Simplex.Messaging.Transport (ATransport (..), TLS, Transport (..), simplexMQVersion)
+import Simplex.Messaging.Transport.Server (loadFingerprint)
 import Simplex.Messaging.Transport.WebSockets (WS)
 import System.Directory (createDirectoryIfMissing, doesDirectoryExist, doesFileExist, removeDirectoryRecursive)
 import System.Exit (exitFailure)
 import System.FilePath (combine)
-import System.IO (BufferMode (..), IOMode (..), hGetLine, withFile, hSetBuffering, stderr, stdout)
+import System.IO (BufferMode (..), IOMode (..), hGetLine, hSetBuffering, stderr, stdout, withFile)
 import System.Process (readCreateProcess, shell)
 import Text.Read (readMaybe)
 

cabal.project

@@ -0,0 +1,6 @@
+packages: .
+
+source-repository-package
+    type: git
+    location: git://github.com/simplex-chat/aeson.git
+    tag: 3eb66f9a68f103b5f1489382aad89f5712a64db7

migrations/README.md

@@ -1,9 +0,0 @@
-# SQLite database migrations
-
-These migrations are [embedded](../src/Simplex/Messaging/Agent/Store/SQLite/Migrations.hs) into the executable and run when SMP agent starts (as a separate executable or as a part of [simplex-chat](https://github.com/simplex-chat/simplex-chat) app).
-
-Migration file names must have a format `YYYYMMDD-name.sql` - they will be executed in the order or lexicographic sorting of the names, the files with any other extension than `.sql` are ignored.
-
-The proposed approach is to minimize the number of migrations and merge them together when possible, to align with the agent releases.
-
-**Please note**: Adding or editing migrations will NOT update the migrations embedded into the executable, unless the [Migrations](../src/Simplex/Messaging/Agent/Store/SQLite/Migrations.hs) module is rebuilt - use `stack build --force-dirty` (in addition to edited files it seems to rebuild the files with TH splices and their dependencies, not all files as with `stack clean`).

package.yaml

@@ -1,5 +1,5 @@
 name: simplexmq
-version: 1.0.1
+version: 1.0.2
 synopsis: SimpleXMQ message broker
 description:  |
   This package includes <./docs/Simplex-Messaging-Server.html server>,
@@ -20,15 +20,14 @@ category: Chat, Network, Web, System, Cryptography
 extra-source-files:
   - README.md
   - CHANGELOG.md
-  - migrations/*.*
 
 dependencies:
-  - aeson == 1.5.*
+  - aeson == 2.0.*
   - ansi-terminal >= 0.10 && < 0.12
   - asn1-encoding == 0.9.*
   - asn1-types == 0.3.*
   - async == 2.2.*
-  - attoparsec == 0.13.*
+  - attoparsec == 0.14.*
   - base >= 4.7 && < 5
   - base64-bytestring >= 1.0 && < 1.3
   - bytestring == 0.10.*
@@ -40,7 +39,6 @@ dependencies:
   - data-default == 0.7.*
   - direct-sqlite == 2.3.*
   - directory == 1.3.*
-  - file-embed >= 0.0.14.0 && <= 0.0.15.0
   - filepath == 1.4.*
   - http-types == 0.12.*
   - generic-random >= 1.3 && < 1.5
@@ -57,7 +55,7 @@ dependencies:
   - template-haskell == 2.16.*
   - text == 1.2.*
   - time == 1.9.*
-  - tls == 1.5.*
+  - tls >= 1.5.7 && < 1.6
   - transformers == 0.5.*
   - unliftio == 0.2.*
   - unliftio-core == 0.2.*

protocol/simplex-messaging.md

@@ -835,7 +835,7 @@ smpVersion = 2*2OCTET ; Word16 version number
 pad = *OCTET
 ```
 
-For TLS 1.3 transport client should assert that `sessionIdentifier` is equal to `tls-unique` channel binding defined in [RFC 5929][14] (TLS Finished message struct); we pass it in `serverHello` block to allow communication over some other transport protocol (possibly, with another channel binding).
+For TLS transport client should assert that `sessionIdentifier` is equal to `tls-unique` channel binding defined in [RFC 5929][14] (TLS Finished message struct); we pass it in `serverHello` block to allow communication over some other transport protocol (possibly, with another channel binding).
 
 [1]: https://en.wikipedia.org/wiki/Man-in-the-middle_attack
 [2]: https://en.wikipedia.org/wiki/End-to-end_encryption
@@ -852,4 +852,4 @@ For TLS 1.3 transport client should assert that `sessionIdentifier` is equal to
 [13]: https://datatracker.ietf.org/doc/html/rfc8446
 [14]: https://datatracker.ietf.org/doc/html/rfc5929#section-3
 [15]: https://www.rfc-editor.org/rfc/rfc8709.html
-[16]: https://nacl.cr.yp.to/box.html
+[16]: https://nacl.cr.yp.to/box.html

scripts/smp-server-digitalocean-droplet/README.md

@@ -1,10 +1,4 @@
-# Server image for DigitalOcean
-
-<!-- TODO tested on a fresh DO droplet and this seems to be wrong - we have to update the image to behave as described.
-
-The current image used for 1-click deployment on DigitalOcean does not contain the source or binary of SMP Server - it downloads the compiled binary of the latest release (rather than a particular release) from GitHub.
-
-The upside is that the new image does not have to be created and approved by DigitalOcean every time when the new release is created. -->
+# SMP server image for DigitalOcean
 
 ## How to build an image
 

scripts/smp-server-digitalocean-droplet/files/opt/simplex/initialize_server.sh

@@ -0,0 +1,51 @@
+#!/bin/bash
+
+# Download latest release
+bin_dir="/opt/simplex/bin"
+binary="$bin_dir/smp-server"
+mkdir -p $bin_dir
+curl -L -o $binary https://github.com/simplex-chat/simplexmq/releases/latest/download/smp-server-ubuntu-20_04-x86-64
+chmod +x $binary
+
+# / Add to PATH
+cat > /etc/profile.d/simplex.sh << EOF
+#!/bin/bash
+
+export PATH="$PATH:$bin_dir"
+
+EOF
+# Add to PATH /
+
+# Source and test PATH
+source /etc/profile.d/simplex.sh
+smp-server --version
+
+# Initialize server
+ip_address=$(curl ifconfig.me)
+smp-server init -l --ip $ip_address
+
+# Server fingerprint
+fingerprint=$(cat /etc/opt/simplex/fingerprint)
+
+# Set up welcome script
+echo "bash /opt/simplex/on_login.sh $fingerprint $ip_address" >> /root/.bashrc
+
+# / Create systemd service for SMP server
+cat > /etc/systemd/system/smp-server.service << EOF
+[Unit]
+Description=SMP server
+
+[Service]
+Type=simple
+ExecStart=/bin/sh -c "exec $binary start >> /var/opt/simplex/smp-server.log 2>&1"
+
+[Install]
+WantedBy=multi-user.target
+
+EOF
+# Create systemd service for SMP server /
+
+# Start systemd service for SMP server
+chmod 644 /etc/systemd/system/smp-server.service
+sudo systemctl enable smp-server
+sudo systemctl start smp-server

scripts/smp-server-digitalocean-droplet/files/opt/simplex/server_bootstrap.sh

@@ -0,0 +1,12 @@
+#!/bin/bash
+
+set -eu
+
+if [[ ! -f /opt/simplex/do_initialize_server ]]; then
+  touch /opt/simplex/do_initialize_server
+elif [[ ! -f /etc/opt/simplex/smp-server.ini ]]; then
+  chmod +x /opt/simplex/initialize_server.sh
+  /opt/simplex/initialize_server.sh
+else
+  echo "SMP server already initialized"
+fi

scripts/smp-server-digitalocean-droplet/scripts/03-init-start.sh

@@ -1,51 +1,23 @@
 #!/bin/bash
 
-# Download latest release
-bin_dir="/opt/simplex/bin"
-binary="$bin_dir/smp-server"
-mkdir -p $bin_dir
-curl -L -o $binary https://github.com/simplex-chat/simplexmq/releases/latest/download/smp-server-ubuntu-20_04-x86-64
-chmod +x $binary
+chmod +x /opt/simplex/server_bootstrap.sh
 
-# / Add to PATH
-cat <<EOT >> /etc/profile.d/simplex.sh
-#!/bin/bash
-
-export PATH="$PATH:$bin_dir"
-
-EOT
-# Add to PATH /
-
-# Source and test PATH
-source /etc/profile.d/simplex.sh
-smp-server --version
-
-# Initialize server
-ip_address=$(curl ifconfig.me)
-smp-server init -l --ip @ip_address
-
-# Server fingerprint
-fingerprint=$(cat /etc/opt/simplex/fingerprint)
-
-# Set up welcome script
-echo "bash /opt/simplex/on_login.sh $fingerprint $ip_address" >> /root/.bashrc
-
-# / Create systemd service
-cat <<EOT >> /etc/systemd/system/smp-server.service
+# / Create systemd service for server bootstrap script
+cat > /etc/systemd/system/server-bootstrap.service << EOF
 [Unit]
-Description=SMP server systemd service
+Description=Server bootstrap script that downloads and initializes SMP server from the latest release
 
 [Service]
-Type=simple
-ExecStart=/bin/sh -c "$binary start"
+Type=oneshot
+ExecStart=/opt/simplex/server_bootstrap.sh
 
 [Install]
 WantedBy=multi-user.target
 
-EOT
-# Create systemd service /
+EOF
+# Create systemd service for server bootstrap script /
 
-# Start systemd service
-chmod 644 /etc/systemd/system/smp-server.service
-sudo systemctl enable smp-server
-sudo systemctl start smp-server
+# Start systemd service for server bootstrap script
+chmod 644 /etc/systemd/system/server-bootstrap.service
+sudo systemctl enable server-bootstrap
+sudo systemctl start server-bootstrap

scripts/smp-server-linode.sh

@@ -52,12 +52,12 @@ curl -L -o $binary https://github.com/simplex-chat/simplexmq/releases/latest/dow
 chmod +x $binary
 
 # / Add to PATH
-cat <<EOT >> /etc/profile.d/simplex.sh
+cat > /etc/profile.d/simplex.sh << EOF
 #!/bin/bash
 
 export PATH="$PATH:$bin_dir"
 
-EOT
+EOF
 # Add to PATH /
 
 # Source and test PATH
@@ -94,13 +94,13 @@ fi
 on_login_script="/opt/simplex/on_login.sh"
 
 # / Welcome script
-cat <<EOT >> $on_login_script
+cat > $on_login_script << EOF
 #!/bin/bash
 
 fingerprint=\$1
 server_address=\$2
 
-cat <<EOF
+cat << EOF2
 ********************************************************************************
 
 SMP server address: smp://\$fingerprint@\$server_address
@@ -111,9 +111,9 @@ All ports are BLOCKED except 22 (SSH), 443 (HTTPS), 5223 (SMP server).
 
 ********************************************************************************
 To stop seeing this message delete line - bash /opt/simplex/on_login.sh - from /root/.bashrc
-EOF
+EOF2
 
-EOT
+EOF
 # Welcome script /
 
 chmod +x $on_login_script
@@ -139,23 +139,23 @@ if [[ -n "$API_TOKEN" ]]; then
   curl \
     -s -H "Content-Type: application/json" \
     -H "Authorization: Bearer $API_TOKEN" \
-    -X PUT -d "{\"tags\":[\"$server_address\",\"#$fingerprint\",\"$version\"]}" \
+    -X PUT -d "{\"tags\":[\"$server_address\",\"$fingerprint\",\"$version\"]}" \
     https://api.linode.com/v4/linode/instances/$LINODE_ID
 fi
 
 # / Create systemd service
-cat <<EOT >> /etc/systemd/system/smp-server.service
+cat > /etc/systemd/system/smp-server.service << EOF
 [Unit]
-Description=SMP server systemd service
+Description=SMP server
 
 [Service]
 Type=simple
-ExecStart=/bin/sh -c "$binary start"
+ExecStart=/bin/sh -c "exec $binary start >> /var/opt/simplex/smp-server.log 2>&1"
 
 [Install]
 WantedBy=multi-user.target
 
-EOT
+EOF
 # Create systemd service /
 
 # Start systemd service

simplexmq.cabal

@@ -5,7 +5,7 @@ cabal-version: 1.12
 -- see: https://github.com/sol/hpack
 
 name:           simplexmq
-version:        1.0.1
+version:        1.0.2
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and
@@ -26,20 +26,24 @@ build-type:     Simple
 extra-source-files:
     README.md
     CHANGELOG.md
-    migrations/20220101_initial.sql
-    migrations/README.md
 
 library
   exposed-modules:
       Simplex.Messaging.Agent
       Simplex.Messaging.Agent.Client
+      Simplex.Messaging.Agent.Env.Postgres
       Simplex.Messaging.Agent.Env.SQLite
       Simplex.Messaging.Agent.Protocol
       Simplex.Messaging.Agent.QueryString
       Simplex.Messaging.Agent.RetryInterval
+      Simplex.Messaging.Agent.Server
       Simplex.Messaging.Agent.Store
+      Simplex.Messaging.Agent.Store.Postgres
+      Simplex.Messaging.Agent.Store.Postgres.Migrations
+      Simplex.Messaging.Agent.Store.Postgres.Migrations.M20220202_initial
       Simplex.Messaging.Agent.Store.SQLite
       Simplex.Messaging.Agent.Store.SQLite.Migrations
+      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220101_initial
       Simplex.Messaging.Client
       Simplex.Messaging.Crypto
       Simplex.Messaging.Crypto.Ratchet
@@ -55,6 +59,8 @@ library
       Simplex.Messaging.Server.QueueStore.STM
       Simplex.Messaging.Server.StoreLog
       Simplex.Messaging.Transport
+      Simplex.Messaging.Transport.Client
+      Simplex.Messaging.Transport.Server
       Simplex.Messaging.Transport.WebSockets
       Simplex.Messaging.Util
       Simplex.Messaging.Version
@@ -65,12 +71,12 @@ library
   ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns
   build-depends:
       QuickCheck ==2.14.*
-    , aeson ==1.5.*
+    , aeson ==2.0.*
     , ansi-terminal >=0.10 && <0.12
     , asn1-encoding ==0.9.*
     , asn1-types ==0.3.*
     , async ==2.2.*
-    , attoparsec ==0.13.*
+    , attoparsec ==0.14.*
     , base >=4.7 && <5
     , base64-bytestring >=1.0 && <1.3
     , bytestring ==0.10.*
@@ -82,7 +88,6 @@ library
     , data-default ==0.7.*
     , direct-sqlite ==2.3.*
     , directory ==1.3.*
-    , file-embed >=0.0.14.0 && <=0.0.15.0
     , filepath ==1.4.*
     , generic-random >=1.3 && <1.5
     , http-types ==0.12.*
@@ -98,7 +103,7 @@ library
     , template-haskell ==2.16.*
     , text ==1.2.*
     , time ==1.9.*
-    , tls ==1.5.*
+    , tls >=1.5.7 && <1.6
     , transformers ==0.5.*
     , unliftio ==0.2.*
     , unliftio-core ==0.2.*
@@ -117,12 +122,12 @@ executable smp-agent
   ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -threaded
   build-depends:
       QuickCheck ==2.14.*
-    , aeson ==1.5.*
+    , aeson ==2.0.*
     , ansi-terminal >=0.10 && <0.12
     , asn1-encoding ==0.9.*
     , asn1-types ==0.3.*
     , async ==2.2.*
-    , attoparsec ==0.13.*
+    , attoparsec ==0.14.*
     , base >=4.7 && <5
     , base64-bytestring >=1.0 && <1.3
     , bytestring ==0.10.*
@@ -134,7 +139,6 @@ executable smp-agent
     , data-default ==0.7.*
     , direct-sqlite ==2.3.*
     , directory ==1.3.*
-    , file-embed >=0.0.14.0 && <=0.0.15.0
     , filepath ==1.4.*
     , generic-random >=1.3 && <1.5
     , http-types ==0.12.*
@@ -151,7 +155,7 @@ executable smp-agent
     , template-haskell ==2.16.*
     , text ==1.2.*
     , time ==1.9.*
-    , tls ==1.5.*
+    , tls >=1.5.7 && <1.6
     , transformers ==0.5.*
     , unliftio ==0.2.*
     , unliftio-core ==0.2.*
@@ -170,12 +174,12 @@ executable smp-server
   ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -threaded
   build-depends:
       QuickCheck ==2.14.*
-    , aeson ==1.5.*
+    , aeson ==2.0.*
     , ansi-terminal >=0.10 && <0.12
     , asn1-encoding ==0.9.*
     , asn1-types ==0.3.*
     , async ==2.2.*
-    , attoparsec ==0.13.*
+    , attoparsec ==0.14.*
     , base >=4.7 && <5
     , base64-bytestring >=1.0 && <1.3
     , bytestring ==0.10.*
@@ -187,7 +191,6 @@ executable smp-server
     , data-default ==0.7.*
     , direct-sqlite ==2.3.*
     , directory ==1.3.*
-    , file-embed >=0.0.14.0 && <=0.0.15.0
     , filepath ==1.4.*
     , generic-random >=1.3 && <1.5
     , http-types ==0.12.*
@@ -207,7 +210,7 @@ executable smp-server
     , template-haskell ==2.16.*
     , text ==1.2.*
     , time ==1.9.*
-    , tls ==1.5.*
+    , tls >=1.5.7 && <1.6
     , transformers ==0.5.*
     , unliftio ==0.2.*
     , unliftio-core ==0.2.*
@@ -239,12 +242,12 @@ test-suite smp-server-test
   build-depends:
       HUnit ==1.6.*
     , QuickCheck ==2.14.*
-    , aeson ==1.5.*
+    , aeson ==2.0.*
     , ansi-terminal >=0.10 && <0.12
     , asn1-encoding ==0.9.*
     , asn1-types ==0.3.*
     , async ==2.2.*
-    , attoparsec ==0.13.*
+    , attoparsec ==0.14.*
     , base >=4.7 && <5
     , base64-bytestring >=1.0 && <1.3
     , bytestring ==0.10.*
@@ -256,7 +259,6 @@ test-suite smp-server-test
     , data-default ==0.7.*
     , direct-sqlite ==2.3.*
     , directory ==1.3.*
-    , file-embed >=0.0.14.0 && <=0.0.15.0
     , filepath ==1.4.*
     , generic-random >=1.3 && <1.5
     , hspec ==2.7.*
@@ -276,7 +278,7 @@ test-suite smp-server-test
     , text ==1.2.*
     , time ==1.9.*
     , timeit ==2.0.*
-    , tls ==1.5.*
+    , tls >=1.5.7 && <1.6
     , transformers ==0.5.*
     , unliftio ==0.2.*
     , unliftio-core ==0.2.*

src/Simplex/Messaging/Agent.hs

@@ -26,11 +26,7 @@
 --
 -- See https://github.com/simplex-chat/simplexmq/blob/master/protocol/agent-protocol.md
 module Simplex.Messaging.Agent
-  ( -- * SMP agent over TCP
-    runSMPAgent,
-    runSMPAgentBlocking,
-
-    -- * queue-based SMP agent
+  ( -- * queue-based SMP agent
     getAgentClient,
     runAgentClient,
 
@@ -51,6 +47,7 @@ module Simplex.Messaging.Agent
     ackMessage,
     suspendConnection,
     deleteConnection,
+    logConnection,
   )
 where
 
@@ -62,7 +59,6 @@ import Control.Monad.Reader
 import Crypto.Random (MonadRandom)
 import Data.Bifunctor (first, second)
 import Data.ByteString.Char8 (ByteString)
-import qualified Data.ByteString.Char8 as B
 import Data.Composition ((.:), (.:.))
 import Data.Functor (($>))
 import Data.List.NonEmpty (NonEmpty (..))
@@ -70,7 +66,6 @@ import qualified Data.List.NonEmpty as L
 import qualified Data.Map.Strict as M
 import Data.Maybe (isJust)
 import qualified Data.Text as T
-import Data.Text.Encoding (decodeUtf8)
 import Data.Time.Clock
 import Data.Time.Clock.System (systemToUTCTime)
 import Database.SQLite.Simple (SQLError)
@@ -87,7 +82,6 @@ import Simplex.Messaging.Encoding
 import Simplex.Messaging.Parsers (parse)
 import Simplex.Messaging.Protocol (MsgBody)
 import qualified Simplex.Messaging.Protocol as SMP
-import Simplex.Messaging.Transport (ATransport (..), TProxy, Transport (..), loadTLSServerParams, runTransportServer, simplexMQVersion)
 import Simplex.Messaging.Util (bshow, liftError, tryError, unlessM)
 import Simplex.Messaging.Version
 import System.Random (randomR)
@@ -95,33 +89,6 @@ import UnliftIO.Async (async, race_)
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
 
--- | Runs an SMP agent as a TCP service using passed configuration.
---
--- See a full agent executable here: https://github.com/simplex-chat/simplexmq/blob/master/apps/smp-agent/Main.hs
-runSMPAgent :: (MonadRandom m, MonadUnliftIO m) => ATransport -> AgentConfig -> m ()
-runSMPAgent t cfg = do
-  started <- newEmptyTMVarIO
-  runSMPAgentBlocking t started cfg
-
--- | Runs an SMP agent as a TCP service using passed configuration with signalling.
---
--- This function uses passed TMVar to signal when the server is ready to accept TCP requests (True)
--- and when it is disconnected from the TCP socket once the server thread is killed (False).
-runSMPAgentBlocking :: (MonadRandom m, MonadUnliftIO m) => ATransport -> TMVar Bool -> AgentConfig -> m ()
-runSMPAgentBlocking (ATransport t) started cfg@AgentConfig {tcpPort, caCertificateFile, certificateFile, privateKeyFile} = do
-  runReaderT (smpAgent t) =<< newSMPAgentEnv cfg
-  where
-    smpAgent :: forall c m'. (Transport c, MonadUnliftIO m', MonadReader Env m') => TProxy c -> m' ()
-    smpAgent _ = do
-      -- tlsServerParams is not in Env to avoid breaking functional API w/t key and certificate generation
-      tlsServerParams <- liftIO $ loadTLSServerParams caCertificateFile certificateFile privateKeyFile
-      runTransportServer started tcpPort tlsServerParams $ \(h :: c) -> do
-        liftIO . putLn h $ "Welcome to SMP agent v" <> B.pack simplexMQVersion
-        c <- getAgentClient
-        logConnection c True
-        race_ (connectClient h c) (runAgentClient c)
-          `E.finally` disconnectAgentClient c
-
 -- | Creates an SMP agent client instance
 getSMPAgentClient :: (MonadRandom m, MonadUnliftIO m) => AgentConfig -> m AgentClient
 getSMPAgentClient cfg = newSMPAgentEnv cfg >>= runReaderT runAgent
@@ -186,9 +153,6 @@ withAgentEnv c = (`runReaderT` agentEnv c)
 getAgentClient :: (MonadUnliftIO m, MonadReader Env m) => m AgentClient
 getAgentClient = ask >>= atomically . newAgentClient
 
-connectClient :: Transport c => MonadUnliftIO m => c -> AgentClient -> m ()
-connectClient h c = race_ (send h c) (receive h c)
-
 logConnection :: MonadUnliftIO m => AgentClient -> Bool -> m ()
 logConnection c connected =
   let event = if connected then "connected to" else "disconnected from"
@@ -198,28 +162,6 @@ logConnection c connected =
 runAgentClient :: (MonadUnliftIO m, MonadReader Env m) => AgentClient -> m ()
 runAgentClient c = race_ (subscriber c) (client c)
 
-receive :: forall c m. (Transport c, MonadUnliftIO m) => c -> AgentClient -> m ()
-receive h c@AgentClient {rcvQ, subQ} = forever $ do
-  (corrId, connId, cmdOrErr) <- tGet SClient h
-  case cmdOrErr of
-    Right cmd -> write rcvQ (corrId, connId, cmd)
-    Left e -> write subQ (corrId, connId, ERR e)
-  where
-    write :: TBQueue (ATransmission p) -> ATransmission p -> m ()
-    write q t = do
-      logClient c "-->" t
-      atomically $ writeTBQueue q t
-
-send :: (Transport c, MonadUnliftIO m) => c -> AgentClient -> m ()
-send h c@AgentClient {subQ} = forever $ do
-  t <- atomically $ readTBQueue subQ
-  tPut h t
-  logClient c "<--" t
-
-logClient :: MonadUnliftIO m => AgentClient -> ByteString -> ATransmission a -> m ()
-logClient AgentClient {clientId} dir (corrId, connId, cmd) = do
-  logInfo . decodeUtf8 $ B.unwords [bshow clientId, dir, "A :", corrId, connId, B.takeWhile (/= ' ') $ serializeCommand cmd]
-
 client :: forall m. (MonadUnliftIO m, MonadReader Env m) => AgentClient -> m ()
 client c@AgentClient {rcvQ, subQ} = forever $ do
   (corrId, connId, cmd) <- atomically $ readTBQueue rcvQ
@@ -458,18 +400,27 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} connId sq = do
     withStore (\st -> E.try $ getPendingMsgData st connId msgId) >>= \case
       Left (e :: E.SomeException) ->
         notify $ MERR mId (INTERNAL $ show e)
-      Right (rq_, (msgType, msgBody)) ->
+      Right (rq_, (msgType, msgBody, internalTs)) ->
         withRetryInterval ri $ \loop ->
           tryError (sendAgentMessage c sq msgBody) >>= \case
             Left e -> do
               case e of
                 SMP SMP.QUOTA -> loop
                 SMP SMP.AUTH -> case msgType of
-                  HELLO_ -> loop
-                  REPLY_ -> notify (ERR e) >> delMsg msgId
-                  A_MSG_ -> notify (MERR mId e) >> delMsg msgId
-                SMP (SMP.CMD _) -> notify (MERR mId e) >> delMsg msgId
-                SMP SMP.LARGE_MSG -> notify (MERR mId e) >> delMsg msgId
+                  HELLO_ -> do
+                    helloTimeout <- asks $ helloTimeout . config
+                    currentTime <- liftIO getCurrentTime
+                    if diffUTCTime currentTime internalTs > helloTimeout
+                      then case rq_ of
+                        -- party initiating connection
+                        Just _ -> notifyDel msgId . ERR $ CONN NOT_AVAILABLE
+                        -- party joining connection
+                        _ -> notifyDel msgId . ERR $ CONN NOT_ACCEPTED
+                      else loop
+                  REPLY_ -> notifyDel msgId $ ERR e
+                  A_MSG_ -> notifyDel msgId $ MERR mId e
+                SMP (SMP.CMD _) -> notifyDel msgId $ MERR mId e
+                SMP SMP.LARGE_MSG -> notifyDel msgId $ MERR mId e
                 SMP {} -> notify (MERR mId e) >> loop
                 _ -> loop
             Right () -> do
@@ -491,6 +442,8 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} connId sq = do
     delMsg msgId = withStore $ \st -> deleteMsg st connId msgId
     notify :: ACommand 'Agent -> m ()
     notify cmd = atomically $ writeTBQueue subQ ("", connId, cmd)
+    notifyDel :: InternalId -> ACommand 'Agent -> m ()
+    notifyDel msgId cmd = notify cmd >> delMsg msgId
 
 ackMessage' :: forall m. AgentMonad m => AgentClient -> ConnId -> AgentMsgId -> m ()
 ackMessage' c connId msgId = do

src/Simplex/Messaging/Agent/Client.hs

@@ -36,6 +36,7 @@ module Simplex.Messaging.Agent.Client
   )
 where
 
+import Control.Concurrent (forkIO)
 import Control.Concurrent.Async (Async, async, uninterruptibleCancel)
 import Control.Concurrent.STM (stateTVar)
 import Control.Logger.Simple
@@ -61,17 +62,19 @@ import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Protocol (QueueId, QueueIdsKeys (..), SndPublicVerifyKey)
 import qualified Simplex.Messaging.Protocol as SMP
-import Simplex.Messaging.Util (bshow, liftEitherError, liftError)
+import Simplex.Messaging.Util (bshow, liftEitherError, liftError, liftIOEither, tryError)
 import Simplex.Messaging.Version
 import UnliftIO.Exception (Exception, IOException)
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
 
+type SMPClientVar = TMVar (Either AgentErrorType SMPClient)
+
 data AgentClient = AgentClient
   { rcvQ :: TBQueue (ATransmission 'Client),
     subQ :: TBQueue (ATransmission 'Agent),
     msgQ :: TBQueue SMPServerTransmission,
-    smpClients :: TVar (Map SMPServer SMPClient),
+    smpClients :: TVar (Map SMPServer SMPClientVar),
     subscrSrvrs :: TVar (Map SMPServer (Map ConnId RcvQueue)),
     subscrConns :: TVar (Map ConnId SMPServer),
     connMsgsQueued :: TVar (Map ConnId Bool),
@@ -118,15 +121,32 @@ instance (MonadUnliftIO m, Exception e) => MonadUnliftIO (ExceptT e m) where
 
 getSMPServerClient :: forall m. AgentMonad m => AgentClient -> SMPServer -> m SMPClient
 getSMPServerClient c@AgentClient {smpClients, msgQ} srv =
-  readTVarIO smpClients
-    >>= maybe newSMPClient return . M.lookup srv
+  atomically getClientVar >>= either newSMPClient waitForSMPClient
   where
-    newSMPClient :: m SMPClient
-    newSMPClient = do
-      smp <- connectClient
-      logInfo . decodeUtf8 $ "Agent connected to " <> showServer srv
-      atomically . modifyTVar smpClients $ M.insert srv smp
-      return smp
+    getClientVar :: STM (Either SMPClientVar SMPClientVar)
+    getClientVar = maybe (Left <$> newClientVar) (pure . Right) . M.lookup srv =<< readTVar smpClients
+
+    newClientVar :: STM SMPClientVar
+    newClientVar = do
+      smpVar <- newEmptyTMVar
+      modifyTVar smpClients $ M.insert srv smpVar
+      pure smpVar
+
+    waitForSMPClient :: TMVar (Either AgentErrorType SMPClient) -> m SMPClient
+    waitForSMPClient = liftIOEither . atomically . readTMVar
+
+    newSMPClient :: TMVar (Either AgentErrorType SMPClient) -> m SMPClient
+    newSMPClient smpVar =
+      tryError connectClient >>= \r -> case r of
+        Right smp -> do
+          logInfo . decodeUtf8 $ "Agent connected to " <> showServer srv
+          atomically $ putTMVar smpVar r
+          pure smp
+        Left e -> do
+          atomically $ do
+            putTMVar smpVar r
+            modifyTVar smpClients $ M.delete srv
+          throwError e
 
     connectClient :: m SMPClient
     connectClient = do
@@ -189,7 +209,12 @@ closeAgentClient c = liftIO $ do
   cancelActions $ smpQueueMsgDeliveries c
 
 closeSMPServerClients :: AgentClient -> IO ()
-closeSMPServerClients c = readTVarIO (smpClients c) >>= mapM_ closeSMPClient
+closeSMPServerClients c = readTVarIO (smpClients c) >>= mapM_ (forkIO . closeClient)
+  where
+    closeClient smpVar =
+      atomically (readTMVar smpVar) >>= \case
+        Right smp -> closeSMPClient smp `E.catch` \(_ :: E.SomeException) -> pure ()
+        _ -> pure ()
 
 cancelActions :: Foldable f => TVar (f (Async ())) -> IO ()
 cancelActions as = readTVarIO as >>= mapM_ uninterruptibleCancel

src/Simplex/Messaging/Agent/Env/Postgres.hs

@@ -0,0 +1,87 @@
+{-# LANGUAGE DataKinds #-}
+{-# LANGUAGE DuplicateRecordFields #-}
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE NumericUnderscores #-}
+{-# OPTIONS_GHC -fno-warn-unticked-promoted-constructors #-}
+
+module Simplex.Messaging.Agent.Env.Postgres
+  ( AgentConfig (..),
+    defaultAgentConfig,
+    Env (..),
+    newSMPAgentEnv,
+  )
+where
+
+import Control.Monad.IO.Unlift
+import Crypto.Random
+import Data.List.NonEmpty (NonEmpty)
+import Data.Time.Clock (NominalDiffTime, nominalDay)
+import Network.Socket
+import Numeric.Natural
+import Simplex.Messaging.Agent.Protocol (SMPServer)
+import Simplex.Messaging.Agent.RetryInterval
+import Simplex.Messaging.Agent.Store.SQLite
+import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations
+import Simplex.Messaging.Client
+import qualified Simplex.Messaging.Crypto as C
+import System.Random (StdGen, newStdGen)
+import UnliftIO.STM
+
+data AgentConfig = AgentConfig
+  { tcpPort :: ServiceName,
+    smpServers :: NonEmpty SMPServer,
+    cmdSignAlg :: C.SignAlg,
+    connIdBytes :: Int,
+    tbqSize :: Natural,
+    dbFile :: FilePath,
+    dbPoolSize :: Int,
+    smpCfg :: SMPClientConfig,
+    reconnectInterval :: RetryInterval,
+    helloTimeout :: NominalDiffTime,
+    caCertificateFile :: FilePath,
+    privateKeyFile :: FilePath,
+    certificateFile :: FilePath
+  }
+
+defaultAgentConfig :: AgentConfig
+defaultAgentConfig =
+  AgentConfig
+    { tcpPort = "5224",
+      smpServers = undefined, -- TODO move it elsewhere?
+      cmdSignAlg = C.SignAlg C.SEd448,
+      connIdBytes = 12,
+      tbqSize = 16,
+      dbFile = "smp-agent.db",
+      dbPoolSize = 4,
+      smpCfg = smpDefaultConfig,
+      reconnectInterval =
+        RetryInterval
+          { initialInterval = second,
+            increaseAfter = 10 * second,
+            maxInterval = 10 * second
+          },
+      helloTimeout = 7 * nominalDay,
+      -- CA certificate private key is not needed for initialization
+      -- ! we do not generate these
+      caCertificateFile = "/etc/opt/simplex-agent/ca.crt",
+      privateKeyFile = "/etc/opt/simplex-agent/agent.key",
+      certificateFile = "/etc/opt/simplex-agent/agent.crt"
+    }
+  where
+    second = 1_000_000
+
+data Env = Env
+  { config :: AgentConfig,
+    store :: SQLiteStore,
+    idsDrg :: TVar ChaChaDRG,
+    clientCounter :: TVar Int,
+    randomServer :: TVar StdGen
+  }
+
+newSMPAgentEnv :: (MonadUnliftIO m, MonadRandom m) => AgentConfig -> m Env
+newSMPAgentEnv cfg = do
+  idsDrg <- newTVarIO =<< drgNew
+  store <- liftIO $ createSQLiteStore (dbFile cfg) (dbPoolSize cfg) Migrations.app
+  clientCounter <- newTVarIO 0
+  randomServer <- newTVarIO =<< liftIO newStdGen
+  return Env {config = cfg, store, idsDrg, clientCounter, randomServer}

src/Simplex/Messaging/Agent/Env/SQLite.hs

@@ -4,11 +4,18 @@
 {-# LANGUAGE NumericUnderscores #-}
 {-# OPTIONS_GHC -fno-warn-unticked-promoted-constructors #-}
 
-module Simplex.Messaging.Agent.Env.SQLite where
+module Simplex.Messaging.Agent.Env.SQLite
+  ( AgentConfig (..),
+    defaultAgentConfig,
+    Env (..),
+    newSMPAgentEnv,
+  )
+where
 
 import Control.Monad.IO.Unlift
 import Crypto.Random
 import Data.List.NonEmpty (NonEmpty)
+import Data.Time.Clock (NominalDiffTime, nominalDay)
 import Network.Socket
 import Numeric.Natural
 import Simplex.Messaging.Agent.Protocol (SMPServer)
@@ -30,14 +37,12 @@ data AgentConfig = AgentConfig
     dbPoolSize :: Int,
     smpCfg :: SMPClientConfig,
     reconnectInterval :: RetryInterval,
+    helloTimeout :: NominalDiffTime,
     caCertificateFile :: FilePath,
     privateKeyFile :: FilePath,
     certificateFile :: FilePath
   }
 
-minute :: Int
-minute = 60_000_000
-
 defaultAgentConfig :: AgentConfig
 defaultAgentConfig =
   AgentConfig
@@ -51,16 +56,19 @@ defaultAgentConfig =
       smpCfg = smpDefaultConfig,
       reconnectInterval =
         RetryInterval
-          { initialInterval = 1_000_000,
-            increaseAfter = 10_000_000,
-            maxInterval = 10_000_000
+          { initialInterval = second,
+            increaseAfter = 10 * second,
+            maxInterval = 10 * second
           },
+      helloTimeout = 7 * nominalDay,
       -- CA certificate private key is not needed for initialization
       -- ! we do not generate these
       caCertificateFile = "/etc/opt/simplex-agent/ca.crt",
       privateKeyFile = "/etc/opt/simplex-agent/agent.key",
       certificateFile = "/etc/opt/simplex-agent/agent.crt"
     }
+  where
+    second = 1_000_000
 
 data Env = Env
   { config :: AgentConfig,

src/Simplex/Messaging/Agent/Protocol.hs

@@ -83,16 +83,10 @@ module Simplex.Messaging.Agent.Protocol
 
     -- * Encode/decode
     serializeCommand,
-    serializeMsgIntegrity,
     connMode,
     connMode',
-    serializeAgentError,
-    serializeSmpErrorType,
     commandP,
     connModeT,
-    msgIntegrityP,
-    agentErrorTypeP,
-    smpErrorTypeP,
     serializeQueueStatus,
     queueStatusT,
     aMessageType,
@@ -108,6 +102,7 @@ where
 import Control.Applicative (optional, (<|>))
 import Control.Monad.IO.Class
 import Data.Aeson (FromJSON (..), ToJSON (..))
+import qualified Data.Aeson as J
 import Data.Attoparsec.ByteString.Char8 (Parser)
 import qualified Data.Attoparsec.ByteString.Char8 as A
 import Data.ByteString.Base64
@@ -131,7 +126,7 @@ import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Crypto.Ratchet (E2ERatchetParams, E2ERatchetParamsUri)
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
-import Simplex.Messaging.Parsers (base64P, parse, parseRead, parseRead1, parseRead2, tsISO8601P)
+import Simplex.Messaging.Parsers
 import Simplex.Messaging.Protocol
   ( ErrorType,
     MsgBody,
@@ -305,7 +300,7 @@ data AgentMsgEnvelope
       }
   | AgentInvitation -- the connInfo in contactInvite is only encrypted with per-queue E2E, not with double ratchet,
       { agentVersion :: Version,
-        connReq :: (ConnectionRequestUri 'CMInvitation),
+        connReq :: ConnectionRequestUri 'CMInvitation,
         connInfo :: ByteString -- this message is only encrypted with per-queue E2E, not with double ratchet,
       }
   deriving (Show)
@@ -618,29 +613,70 @@ queueStatusT = \case
 type AgentMsgId = Int64
 
 -- | Result of received message integrity validation.
-data MsgIntegrity = MsgOk | MsgError MsgErrorType
-  deriving (Eq, Show)
+data MsgIntegrity = MsgOk | MsgError {errorInfo :: MsgErrorType}
+  deriving (Eq, Show, Generic)
+
+instance StrEncoding MsgIntegrity where
+  strP = "OK" $> MsgOk <|> "ERR " *> (MsgError <$> strP)
+  strEncode = \case
+    MsgOk -> "OK"
+    MsgError e -> "ERR" <> strEncode e
+
+instance ToJSON MsgIntegrity where
+  toJSON = J.genericToJSON $ sumTypeJSON fstToLower
+  toEncoding = J.genericToEncoding $ sumTypeJSON fstToLower
+
+instance FromJSON MsgIntegrity where
+  parseJSON = J.genericParseJSON $ sumTypeJSON fstToLower
 
 -- | Error of message integrity validation.
-data MsgErrorType = MsgSkipped AgentMsgId AgentMsgId | MsgBadId AgentMsgId | MsgBadHash | MsgDuplicate
-  deriving (Eq, Show)
+data MsgErrorType
+  = MsgSkipped {fromMsgId :: AgentMsgId, toMsgId :: AgentMsgId}
+  | MsgBadId {msgId :: AgentMsgId}
+  | MsgBadHash
+  | MsgDuplicate
+  deriving (Eq, Show, Generic)
+
+instance StrEncoding MsgErrorType where
+  strP =
+    "ID " *> (MsgBadId <$> A.decimal)
+      <|> "IDS " *> (MsgSkipped <$> A.decimal <* A.space <*> A.decimal)
+      <|> "HASH" $> MsgBadHash
+      <|> "DUPLICATE" $> MsgDuplicate
+  strEncode = \case
+    MsgSkipped fromMsgId toMsgId ->
+      B.unwords ["NO_ID", bshow fromMsgId, bshow toMsgId]
+    MsgBadId aMsgId -> "ID " <> bshow aMsgId
+    MsgBadHash -> "HASH"
+    MsgDuplicate -> "DUPLICATE"
+
+instance ToJSON MsgErrorType where
+  toJSON = J.genericToJSON $ sumTypeJSON fstToLower
+  toEncoding = J.genericToEncoding $ sumTypeJSON fstToLower
+
+instance FromJSON MsgErrorType where
+  parseJSON = J.genericParseJSON $ sumTypeJSON fstToLower
 
 -- | Error type used in errors sent to agent clients.
 data AgentErrorType
   = -- | command or response error
-    CMD CommandErrorType
+    CMD {cmdErr :: CommandErrorType}
   | -- | connection errors
-    CONN ConnectionErrorType
+    CONN {connErr :: ConnectionErrorType}
   | -- | SMP protocol errors forwarded to agent clients
-    SMP ErrorType
+    SMP {smpErr :: ErrorType}
   | -- | SMP server errors
-    BROKER BrokerErrorType
+    BROKER {brokerErr :: BrokerErrorType}
   | -- | errors of other agents
-    AGENT SMPAgentError
+    AGENT {agentErr :: SMPAgentError}
   | -- | agent implementation or dependency errors
-    INTERNAL String
+    INTERNAL {internalErr :: String}
   deriving (Eq, Generic, Read, Show, Exception)
 
+instance ToJSON AgentErrorType where
+  toJSON = J.genericToJSON $ sumTypeJSON id
+  toEncoding = J.genericToEncoding $ sumTypeJSON id
+
 -- | SMP agent protocol command or response error.
 data CommandErrorType
   = -- | command is prohibited in this context
@@ -655,6 +691,10 @@ data CommandErrorType
     LARGE
   deriving (Eq, Generic, Read, Show, Exception)
 
+instance ToJSON CommandErrorType where
+  toJSON = J.genericToJSON $ sumTypeJSON id
+  toEncoding = J.genericToEncoding $ sumTypeJSON id
+
 -- | Connection error.
 data ConnectionErrorType
   = -- | connection is not in the database
@@ -663,22 +703,34 @@ data ConnectionErrorType
     DUPLICATE
   | -- | connection is simplex, but operation requires another queue
     SIMPLEX
+  | -- | connection not accepted on join HELLO after timeout
+    NOT_ACCEPTED
+  | -- | connection not available on reply HELLO after timeout
+    NOT_AVAILABLE
   deriving (Eq, Generic, Read, Show, Exception)
 
+instance ToJSON ConnectionErrorType where
+  toJSON = J.genericToJSON $ sumTypeJSON id
+  toEncoding = J.genericToEncoding $ sumTypeJSON id
+
 -- | SMP server errors.
 data BrokerErrorType
   = -- | invalid server response (failed to parse)
-    RESPONSE ErrorType
+    RESPONSE {smpErr :: ErrorType}
   | -- | unexpected response
     UNEXPECTED
   | -- | network error
     NETWORK
   | -- | handshake or other transport error
-    TRANSPORT TransportError
+    TRANSPORT {transportErr :: TransportError}
   | -- | command response timeout
     TIMEOUT
   deriving (Eq, Generic, Read, Show, Exception)
 
+instance ToJSON BrokerErrorType where
+  toJSON = J.genericToJSON $ sumTypeJSON id
+  toEncoding = J.genericToEncoding $ sumTypeJSON id
+
 -- | Errors of another SMP agent.
 -- TODO encode/decode without A prefix
 data SMPAgentError
@@ -692,6 +744,30 @@ data SMPAgentError
     A_ENCRYPTION
   deriving (Eq, Generic, Read, Show, Exception)
 
+instance ToJSON SMPAgentError where
+  toJSON = J.genericToJSON $ sumTypeJSON id
+  toEncoding = J.genericToEncoding $ sumTypeJSON id
+
+instance StrEncoding AgentErrorType where
+  strP =
+    "CMD " *> (CMD <$> parseRead1)
+      <|> "CONN " *> (CONN <$> parseRead1)
+      <|> "SMP " *> (SMP <$> strP)
+      <|> "BROKER RESPONSE " *> (BROKER . RESPONSE <$> strP)
+      <|> "BROKER TRANSPORT " *> (BROKER . TRANSPORT <$> transportErrorP)
+      <|> "BROKER " *> (BROKER <$> parseRead1)
+      <|> "AGENT " *> (AGENT <$> parseRead1)
+      <|> "INTERNAL " *> (INTERNAL <$> parseRead A.takeByteString)
+  strEncode = \case
+    CMD e -> "CMD " <> bshow e
+    CONN e -> "CONN " <> bshow e
+    SMP e -> "SMP " <> strEncode e
+    BROKER (RESPONSE e) -> "BROKER RESPONSE " <> strEncode e
+    BROKER (TRANSPORT e) -> "BROKER TRANSPORT " <> serializeTransportError e
+    BROKER e -> "BROKER " <> bshow e
+    AGENT e -> "AGENT " <> bshow e
+    INTERNAL e -> "INTERNAL " <> bshow e
+
 instance Arbitrary AgentErrorType where arbitrary = genericArbitraryU
 
 instance Arbitrary CommandErrorType where arbitrary = genericArbitraryU
@@ -742,27 +818,17 @@ commandP =
     sendCmd = ACmd SClient . SEND <$> A.takeByteString
     msgIdResp = ACmd SAgent . MID <$> A.decimal
     sentResp = ACmd SAgent . SENT <$> A.decimal
-    msgErrResp = ACmd SAgent .: MERR <$> A.decimal <* A.space <*> agentErrorTypeP
+    msgErrResp = ACmd SAgent .: MERR <$> A.decimal <* A.space <*> strP
     message = ACmd SAgent .: MSG <$> msgMetaP <* A.space <*> A.takeByteString
     ackCmd = ACmd SClient . ACK <$> A.decimal
     msgMetaP = do
-      integrity <- msgIntegrityP
+      integrity <- strP
       recipient <- " R=" *> partyMeta A.decimal
       broker <- " B=" *> partyMeta base64P
       sndMsgId <- " S=" *> A.decimal
       pure MsgMeta {integrity, recipient, broker, sndMsgId}
     partyMeta idParser = (,) <$> idParser <* A.char ',' <*> tsISO8601P
-    agentError = ACmd SAgent . ERR <$> agentErrorTypeP
-
--- | Message integrity validation result parser.
-msgIntegrityP :: Parser MsgIntegrity
-msgIntegrityP = "OK" $> MsgOk <|> "ERR " *> (MsgError <$> msgErrorType)
-  where
-    msgErrorType =
-      "ID " *> (MsgBadId <$> A.decimal)
-        <|> "IDS " *> (MsgSkipped <$> A.decimal <* A.space <*> A.decimal)
-        <|> "HASH" $> MsgBadHash
-        <|> "DUPLICATE" $> MsgDuplicate
+    agentError = ACmd SAgent . ERR <$> strP
 
 parseCommand :: ByteString -> Either AgentErrorType ACmd
 parseCommand = parse commandP $ CMD SYNTAX
@@ -786,13 +852,13 @@ serializeCommand = \case
   SEND msgBody -> "SEND " <> serializeBinary msgBody
   MID mId -> "MID " <> bshow mId
   SENT mId -> "SENT " <> bshow mId
-  MERR mId e -> B.unwords ["MERR", bshow mId, serializeAgentError e]
+  MERR mId e -> B.unwords ["MERR", bshow mId, strEncode e]
   MSG msgMeta msgBody -> B.unwords ["MSG", serializeMsgMeta msgMeta, serializeBinary msgBody]
   ACK mId -> "ACK " <> bshow mId
   OFF -> "OFF"
   DEL -> "DEL"
   CON -> "CON"
-  ERR e -> "ERR " <> serializeAgentError e
+  ERR e -> "ERR " <> strEncode e
   OK -> "OK"
   where
     showTs :: UTCTime -> ByteString
@@ -800,49 +866,12 @@ serializeCommand = \case
     serializeMsgMeta :: MsgMeta -> ByteString
     serializeMsgMeta MsgMeta {integrity, recipient = (rmId, rTs), broker = (bmId, bTs), sndMsgId} =
       B.unwords
-        [ serializeMsgIntegrity integrity,
+        [ strEncode integrity,
           "R=" <> bshow rmId <> "," <> showTs rTs,
           "B=" <> encode bmId <> "," <> showTs bTs,
           "S=" <> bshow sndMsgId
         ]
 
--- | Serialize message integrity validation result.
-serializeMsgIntegrity :: MsgIntegrity -> ByteString
-serializeMsgIntegrity = \case
-  MsgOk -> "OK"
-  MsgError e ->
-    "ERR " <> case e of
-      MsgSkipped fromMsgId toMsgId ->
-        B.unwords ["NO_ID", bshow fromMsgId, bshow toMsgId]
-      MsgBadId aMsgId -> "ID " <> bshow aMsgId
-      MsgBadHash -> "HASH"
-      MsgDuplicate -> "DUPLICATE"
-
--- | SMP agent protocol error parser.
-agentErrorTypeP :: Parser AgentErrorType
-agentErrorTypeP =
-  "SMP " *> (SMP <$> smpErrorTypeP)
-    <|> "BROKER RESPONSE " *> (BROKER . RESPONSE <$> smpErrorTypeP)
-    <|> "BROKER TRANSPORT " *> (BROKER . TRANSPORT <$> transportErrorP)
-    <|> "INTERNAL " *> (INTERNAL <$> parseRead A.takeByteString)
-    <|> parseRead2
-
--- | Serialize SMP agent protocol error.
-serializeAgentError :: AgentErrorType -> ByteString
-serializeAgentError = \case
-  SMP e -> "SMP " <> serializeSmpErrorType e
-  BROKER (RESPONSE e) -> "BROKER RESPONSE " <> serializeSmpErrorType e
-  BROKER (TRANSPORT e) -> "BROKER TRANSPORT " <> serializeTransportError e
-  e -> bshow e
-
--- | SMP error parser.
-smpErrorTypeP :: Parser ErrorType
-smpErrorTypeP = "CMD " *> (SMP.CMD <$> parseRead1) <|> parseRead1
-
--- | Serialize SMP error.
-serializeSmpErrorType :: ErrorType -> ByteString
-serializeSmpErrorType = bshow
-
 serializeBinary :: ByteString -> ByteString
 serializeBinary body = bshow (B.length body) <> "\n" <> body
 

src/Simplex/Messaging/Agent/Server.hs

@@ -0,0 +1,81 @@
+{-# LANGUAGE FlexibleContexts #-}
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+
+module Simplex.Messaging.Agent.Server
+  ( -- * SMP agent over TCP
+    runSMPAgent,
+    runSMPAgentBlocking,
+  )
+where
+
+import Control.Logger.Simple (logInfo)
+import Control.Monad.Except
+import Control.Monad.IO.Unlift (MonadUnliftIO)
+import Control.Monad.Reader
+import Crypto.Random (MonadRandom)
+import Data.ByteString.Char8 (ByteString)
+import qualified Data.ByteString.Char8 as B
+import Data.Text.Encoding (decodeUtf8)
+import Simplex.Messaging.Agent
+import Simplex.Messaging.Agent.Env.SQLite
+import Simplex.Messaging.Agent.Protocol
+import Simplex.Messaging.Transport (ATransport (..), TProxy, Transport (..), simplexMQVersion)
+import Simplex.Messaging.Transport.Server (loadTLSServerParams, runTransportServer)
+import Simplex.Messaging.Util (bshow)
+import UnliftIO.Async (race_)
+import qualified UnliftIO.Exception as E
+import UnliftIO.STM
+
+-- | Runs an SMP agent as a TCP service using passed configuration.
+--
+-- See a full agent executable here: https://github.com/simplex-chat/simplexmq/blob/master/apps/smp-agent/Main.hs
+runSMPAgent :: (MonadRandom m, MonadUnliftIO m) => ATransport -> AgentConfig -> m ()
+runSMPAgent t cfg = do
+  started <- newEmptyTMVarIO
+  runSMPAgentBlocking t started cfg
+
+-- | Runs an SMP agent as a TCP service using passed configuration with signalling.
+--
+-- This function uses passed TMVar to signal when the server is ready to accept TCP requests (True)
+-- and when it is disconnected from the TCP socket once the server thread is killed (False).
+runSMPAgentBlocking :: (MonadRandom m, MonadUnliftIO m) => ATransport -> TMVar Bool -> AgentConfig -> m ()
+runSMPAgentBlocking (ATransport t) started cfg@AgentConfig {tcpPort, caCertificateFile, certificateFile, privateKeyFile} = do
+  runReaderT (smpAgent t) =<< newSMPAgentEnv cfg
+  where
+    smpAgent :: forall c m'. (Transport c, MonadUnliftIO m', MonadReader Env m') => TProxy c -> m' ()
+    smpAgent _ = do
+      -- tlsServerParams is not in Env to avoid breaking functional API w/t key and certificate generation
+      tlsServerParams <- liftIO $ loadTLSServerParams caCertificateFile certificateFile privateKeyFile
+      runTransportServer started tcpPort tlsServerParams $ \(h :: c) -> do
+        liftIO . putLn h $ "Welcome to SMP agent v" <> B.pack simplexMQVersion
+        c <- getAgentClient
+        logConnection c True
+        race_ (connectClient h c) (runAgentClient c)
+          `E.finally` disconnectAgentClient c
+
+connectClient :: Transport c => MonadUnliftIO m => c -> AgentClient -> m ()
+connectClient h c = race_ (send h c) (receive h c)
+
+receive :: forall c m. (Transport c, MonadUnliftIO m) => c -> AgentClient -> m ()
+receive h c@AgentClient {rcvQ, subQ} = forever $ do
+  (corrId, connId, cmdOrErr) <- tGet SClient h
+  case cmdOrErr of
+    Right cmd -> write rcvQ (corrId, connId, cmd)
+    Left e -> write subQ (corrId, connId, ERR e)
+  where
+    write :: TBQueue (ATransmission p) -> ATransmission p -> m ()
+    write q t = do
+      logClient c "-->" t
+      atomically $ writeTBQueue q t
+
+send :: (Transport c, MonadUnliftIO m) => c -> AgentClient -> m ()
+send h c@AgentClient {subQ} = forever $ do
+  t <- atomically $ readTBQueue subQ
+  tPut h t
+  logClient c "<--" t
+
+logClient :: MonadUnliftIO m => AgentClient -> ByteString -> ATransmission a -> m ()
+logClient AgentClient {clientId} dir (corrId, connId, cmd) = do
+  logInfo . decodeUtf8 $ B.unwords [bshow clientId, dir, "A :", corrId, connId, B.takeWhile (/= ' ') $ serializeCommand cmd]

src/Simplex/Messaging/Agent/Store.hs

@@ -62,7 +62,7 @@ class Monad m => MonadAgentStore s m where
   createRcvMsg :: s -> ConnId -> RcvMsgData -> m ()
   updateSndIds :: s -> ConnId -> m (InternalId, InternalSndId, PrevSndMsgHash)
   createSndMsg :: s -> ConnId -> SndMsgData -> m ()
-  getPendingMsgData :: s -> ConnId -> InternalId -> m (Maybe RcvQueue, (AMsgType, MsgBody))
+  getPendingMsgData :: s -> ConnId -> InternalId -> m (Maybe RcvQueue, (AMsgType, MsgBody, InternalTs))
   getPendingMsgs :: s -> ConnId -> m [InternalId]
   checkRcvMsg :: s -> ConnId -> InternalId -> m ()
   deleteMsg :: s -> ConnId -> InternalId -> m ()

src/Simplex/Messaging/Agent/Store/Postgres.hs

@@ -0,0 +1,953 @@
+{-# LANGUAGE DataKinds #-}
+{-# LANGUAGE DuplicateRecordFields #-}
+{-# LANGUAGE FlexibleContexts #-}
+{-# LANGUAGE FlexibleInstances #-}
+{-# LANGUAGE GADTs #-}
+{-# LANGUAGE InstanceSigs #-}
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE MultiParamTypeClasses #-}
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE NumericUnderscores #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE QuasiQuotes #-}
+{-# LANGUAGE RecordWildCards #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TupleSections #-}
+{-# LANGUAGE UndecidableInstances #-}
+{-# OPTIONS_GHC -fno-warn-orphans #-}
+
+module Simplex.Messaging.Agent.Store.Postgres
+  ( SQLiteStore (..),
+    createSQLiteStore,
+    connectSQLiteStore,
+    withConnection,
+    withTransaction,
+    fromTextField_,
+    firstRow,
+  )
+where
+
+import Control.Concurrent (threadDelay)
+import Control.Concurrent.STM
+import Control.Exception (bracket)
+import Control.Monad.Except
+import Control.Monad.IO.Unlift (MonadUnliftIO)
+import Crypto.Random (ChaChaDRG, randomBytesGenerate)
+import Data.Bifunctor (second)
+import Data.ByteString (ByteString)
+import qualified Data.ByteString.Base64.URL as U
+import Data.Char (toLower)
+import Data.Functor (($>))
+import Data.List (find, foldl')
+import qualified Data.Map.Strict as M
+import Data.Text (Text)
+import qualified Data.Text as T
+import Data.Text.Encoding (decodeLatin1)
+import Database.SQLite.Simple (FromRow, NamedParam (..), Only (..), SQLData (..), SQLError, ToRow, field)
+import qualified Database.SQLite.Simple as DB
+import Database.SQLite.Simple.FromField
+import Database.SQLite.Simple.Internal (Field (..))
+import Database.SQLite.Simple.Ok (Ok (Ok))
+import Database.SQLite.Simple.QQ (sql)
+import Database.SQLite.Simple.ToField (ToField (..))
+import Simplex.Messaging.Agent.Protocol
+import Simplex.Messaging.Agent.Store
+import Simplex.Messaging.Agent.Store.SQLite.Migrations (Migration)
+import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations
+import qualified Simplex.Messaging.Crypto as C
+import Simplex.Messaging.Crypto.Ratchet (RatchetX448, SkippedMsgDiff (..), SkippedMsgKeys)
+import Simplex.Messaging.Encoding
+import Simplex.Messaging.Encoding.String
+import Simplex.Messaging.Parsers (blobFieldParser)
+import Simplex.Messaging.Protocol (MsgBody)
+import qualified Simplex.Messaging.Protocol as SMP
+import Simplex.Messaging.Util (bshow, liftIOEither)
+import System.Directory (copyFile, createDirectoryIfMissing, doesFileExist)
+import System.Exit (exitFailure)
+import System.FilePath (takeDirectory)
+import System.IO (hFlush, stdout)
+import qualified UnliftIO.Exception as E
+
+-- * SQLite Store implementation
+
+data SQLiteStore = SQLiteStore
+  { dbFilePath :: FilePath,
+    dbConnPool :: TBQueue DB.Connection,
+    dbNew :: Bool
+  }
+
+createSQLiteStore :: FilePath -> Int -> [Migration] -> IO SQLiteStore
+createSQLiteStore dbFilePath poolSize migrations = do
+  let dbDir = takeDirectory dbFilePath
+  createDirectoryIfMissing False dbDir
+  st <- connectSQLiteStore dbFilePath poolSize
+  checkThreadsafe st
+  migrateSchema st migrations
+  pure st
+
+checkThreadsafe :: SQLiteStore -> IO ()
+checkThreadsafe st = withConnection st $ \db -> do
+  compileOptions <- DB.query_ db "pragma COMPILE_OPTIONS;" :: IO [[Text]]
+  let threadsafeOption = find (T.isPrefixOf "THREADSAFE=") (concat compileOptions)
+  case threadsafeOption of
+    Just "THREADSAFE=0" -> confirmOrExit "SQLite compiled with non-threadsafe code."
+    Nothing -> putStrLn "Warning: SQLite THREADSAFE compile option not found"
+    _ -> return ()
+
+migrateSchema :: SQLiteStore -> [Migration] -> IO ()
+migrateSchema st migrations = withConnection st $ \db -> do
+  Migrations.initialize db
+  Migrations.get db migrations >>= \case
+    Left e -> confirmOrExit $ "Database error: " <> e
+    Right [] -> pure ()
+    Right ms -> do
+      unless (dbNew st) $ do
+        confirmOrExit "The app has a newer version than the database - it will be backed up and upgraded."
+        let f = dbFilePath st
+        copyFile f (f <> ".bak")
+      Migrations.run db ms
+
+confirmOrExit :: String -> IO ()
+confirmOrExit s = do
+  putStrLn s
+  putStr "Continue (y/N): "
+  hFlush stdout
+  ok <- getLine
+  when (map toLower ok /= "y") exitFailure
+
+connectSQLiteStore :: FilePath -> Int -> IO SQLiteStore
+connectSQLiteStore dbFilePath poolSize = do
+  dbNew <- not <$> doesFileExist dbFilePath
+  dbConnPool <- newTBQueueIO $ toEnum poolSize
+  replicateM_ poolSize $
+    connectDB dbFilePath >>= atomically . writeTBQueue dbConnPool
+  pure SQLiteStore {dbFilePath, dbConnPool, dbNew}
+
+connectDB :: FilePath -> IO DB.Connection
+connectDB path = do
+  dbConn <- DB.open path
+  DB.execute_ dbConn "PRAGMA foreign_keys = ON; PRAGMA journal_mode = WAL;"
+  pure dbConn
+
+checkConstraint :: StoreError -> IO (Either StoreError a) -> IO (Either StoreError a)
+checkConstraint err action = action `E.catch` (pure . Left . handleSQLError err)
+
+handleSQLError :: StoreError -> SQLError -> StoreError
+handleSQLError err e
+  | DB.sqlError e == DB.ErrorConstraint = err
+  | otherwise = SEInternal $ bshow e
+
+withConnection :: SQLiteStore -> (DB.Connection -> IO a) -> IO a
+withConnection SQLiteStore {dbConnPool} =
+  bracket
+    (atomically $ readTBQueue dbConnPool)
+    (atomically . writeTBQueue dbConnPool)
+
+withTransaction :: forall a. SQLiteStore -> (DB.Connection -> IO a) -> IO a
+withTransaction st action = withConnection st $ loop 100 100_000
+  where
+    loop :: Int -> Int -> DB.Connection -> IO a
+    loop t tLim db =
+      DB.withImmediateTransaction db (action db) `E.catch` \(e :: SQLError) ->
+        if tLim > t && DB.sqlError e == DB.ErrorBusy
+          then do
+            threadDelay t
+            loop (t * 9 `div` 8) (tLim - t) db
+          else E.throwIO e
+
+createConn_ ::
+  (MonadUnliftIO m, MonadError StoreError m) =>
+  SQLiteStore ->
+  TVar ChaChaDRG ->
+  ConnData ->
+  (DB.Connection -> ByteString -> IO ()) ->
+  m ByteString
+createConn_ st gVar cData create =
+  liftIOEither . checkConstraint SEConnDuplicate . withTransaction st $ \db ->
+    case cData of
+      ConnData {connId = ""} -> createWithRandomId gVar $ create db
+      ConnData {connId} -> create db connId $> Right connId
+
+instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteStore m where
+  createRcvConn :: SQLiteStore -> TVar ChaChaDRG -> ConnData -> RcvQueue -> SConnectionMode c -> m ConnId
+  createRcvConn st gVar cData q@RcvQueue {server} cMode =
+    createConn_ st gVar cData $ \db connId -> do
+      upsertServer_ db server
+      DB.execute db "INSERT INTO connections (conn_id, conn_mode) VALUES (?, ?)" (connId, cMode)
+      insertRcvQueue_ db connId q
+
+  createSndConn :: SQLiteStore -> TVar ChaChaDRG -> ConnData -> SndQueue -> m ConnId
+  createSndConn st gVar cData q@SndQueue {server} =
+    createConn_ st gVar cData $ \db connId -> do
+      upsertServer_ db server
+      DB.execute db "INSERT INTO connections (conn_id, conn_mode) VALUES (?, ?)" (connId, SCMInvitation)
+      insertSndQueue_ db connId q
+
+  getConn :: SQLiteStore -> ConnId -> m SomeConn
+  getConn st connId =
+    liftIOEither . withTransaction st $ \db ->
+      getConn_ db connId
+
+  getRcvConn :: SQLiteStore -> SMPServer -> SMP.RecipientId -> m SomeConn
+  getRcvConn st SMPServer {host, port} rcvId =
+    liftIOEither . withTransaction st $ \db ->
+      DB.queryNamed
+        db
+        [sql|
+          SELECT q.conn_id
+          FROM rcv_queues q
+          WHERE q.host = :host AND q.port = :port AND q.rcv_id = :rcv_id;
+        |]
+        [":host" := host, ":port" := port, ":rcv_id" := rcvId]
+        >>= \case
+          [Only connId] -> getConn_ db connId
+          _ -> pure $ Left SEConnNotFound
+
+  deleteConn :: SQLiteStore -> ConnId -> m ()
+  deleteConn st connId =
+    liftIO . withTransaction st $ \db ->
+      DB.executeNamed
+        db
+        "DELETE FROM connections WHERE conn_id = :conn_id;"
+        [":conn_id" := connId]
+
+  upgradeRcvConnToDuplex :: SQLiteStore -> ConnId -> SndQueue -> m ()
+  upgradeRcvConnToDuplex st connId sq@SndQueue {server} =
+    liftIOEither . withTransaction st $ \db ->
+      getConn_ db connId >>= \case
+        Right (SomeConn _ RcvConnection {}) -> do
+          upsertServer_ db server
+          insertSndQueue_ db connId sq
+          pure $ Right ()
+        Right (SomeConn c _) -> pure . Left . SEBadConnType $ connType c
+        _ -> pure $ Left SEConnNotFound
+
+  upgradeSndConnToDuplex :: SQLiteStore -> ConnId -> RcvQueue -> m ()
+  upgradeSndConnToDuplex st connId rq@RcvQueue {server} =
+    liftIOEither . withTransaction st $ \db ->
+      getConn_ db connId >>= \case
+        Right (SomeConn _ SndConnection {}) -> do
+          upsertServer_ db server
+          insertRcvQueue_ db connId rq
+          pure $ Right ()
+        Right (SomeConn c _) -> pure . Left . SEBadConnType $ connType c
+        _ -> pure $ Left SEConnNotFound
+
+  setRcvQueueStatus :: SQLiteStore -> RcvQueue -> QueueStatus -> m ()
+  setRcvQueueStatus st RcvQueue {rcvId, server = SMPServer {host, port}} status =
+    -- ? throw error if queue does not exist?
+    liftIO . withTransaction st $ \db ->
+      DB.executeNamed
+        db
+        [sql|
+          UPDATE rcv_queues
+          SET status = :status
+          WHERE host = :host AND port = :port AND rcv_id = :rcv_id;
+        |]
+        [":status" := status, ":host" := host, ":port" := port, ":rcv_id" := rcvId]
+
+  setRcvQueueConfirmedE2E :: SQLiteStore -> RcvQueue -> C.DhSecretX25519 -> m ()
+  setRcvQueueConfirmedE2E st RcvQueue {rcvId, server = SMPServer {host, port}} e2eDhSecret =
+    liftIO . withTransaction st $ \db ->
+      DB.executeNamed
+        db
+        [sql|
+          UPDATE rcv_queues
+          SET e2e_dh_secret = :e2e_dh_secret,
+              status = :status
+          WHERE host = :host AND port = :port AND rcv_id = :rcv_id
+        |]
+        [ ":status" := Confirmed,
+          ":e2e_dh_secret" := e2eDhSecret,
+          ":host" := host,
+          ":port" := port,
+          ":rcv_id" := rcvId
+        ]
+
+  setSndQueueStatus :: SQLiteStore -> SndQueue -> QueueStatus -> m ()
+  setSndQueueStatus st SndQueue {sndId, server = SMPServer {host, port}} status =
+    -- ? throw error if queue does not exist?
+    liftIO . withTransaction st $ \db ->
+      DB.executeNamed
+        db
+        [sql|
+          UPDATE snd_queues
+          SET status = :status
+          WHERE host = :host AND port = :port AND snd_id = :snd_id;
+        |]
+        [":status" := status, ":host" := host, ":port" := port, ":snd_id" := sndId]
+
+  createConfirmation :: SQLiteStore -> TVar ChaChaDRG -> NewConfirmation -> m ConfirmationId
+  createConfirmation st gVar NewConfirmation {connId, senderConf = SMPConfirmation {senderKey, e2ePubKey, connInfo}, ratchetState} =
+    liftIOEither . withTransaction st $ \db ->
+      createWithRandomId gVar $ \confirmationId ->
+        DB.execute
+          db
+          [sql|
+            INSERT INTO conn_confirmations
+            (confirmation_id, conn_id, sender_key, e2e_snd_pub_key, ratchet_state, sender_conn_info, accepted) VALUES (?, ?, ?, ?, ?, ?, 0);
+          |]
+          (confirmationId, connId, senderKey, e2ePubKey, ratchetState, connInfo)
+
+  acceptConfirmation :: SQLiteStore -> ConfirmationId -> ConnInfo -> m AcceptedConfirmation
+  acceptConfirmation st confirmationId ownConnInfo =
+    liftIOEither . withTransaction st $ \db -> do
+      DB.executeNamed
+        db
+        [sql|
+          UPDATE conn_confirmations
+          SET accepted = 1,
+              own_conn_info = :own_conn_info
+          WHERE confirmation_id = :confirmation_id;
+        |]
+        [ ":own_conn_info" := ownConnInfo,
+          ":confirmation_id" := confirmationId
+        ]
+      firstRow confirmation SEConfirmationNotFound $
+        DB.query
+          db
+          [sql|
+            SELECT conn_id, sender_key, e2e_snd_pub_key, ratchet_state, sender_conn_info
+            FROM conn_confirmations
+            WHERE confirmation_id = ?;
+          |]
+          (Only confirmationId)
+    where
+      confirmation (connId, senderKey, e2ePubKey, ratchetState, connInfo) =
+        AcceptedConfirmation
+          { confirmationId,
+            connId,
+            senderConf = SMPConfirmation {senderKey, e2ePubKey, connInfo},
+            ratchetState,
+            ownConnInfo
+          }
+
+  getAcceptedConfirmation :: SQLiteStore -> ConnId -> m AcceptedConfirmation
+  getAcceptedConfirmation st connId =
+    liftIOEither . withTransaction st $ \db ->
+      firstRow confirmation SEConfirmationNotFound $
+        DB.query
+          db
+          [sql|
+            SELECT confirmation_id, sender_key, e2e_snd_pub_key, ratchet_state, sender_conn_info, own_conn_info
+            FROM conn_confirmations
+            WHERE conn_id = ? AND accepted = 1;
+          |]
+          (Only connId)
+    where
+      confirmation (confirmationId, senderKey, e2ePubKey, ratchetState, connInfo, ownConnInfo) =
+        AcceptedConfirmation
+          { confirmationId,
+            connId,
+            senderConf = SMPConfirmation {senderKey, e2ePubKey, connInfo},
+            ratchetState,
+            ownConnInfo
+          }
+
+  removeConfirmations :: SQLiteStore -> ConnId -> m ()
+  removeConfirmations st connId =
+    liftIO . withTransaction st $ \db ->
+      DB.executeNamed
+        db
+        [sql|
+          DELETE FROM conn_confirmations
+          WHERE conn_id = :conn_id;
+        |]
+        [":conn_id" := connId]
+
+  createInvitation :: SQLiteStore -> TVar ChaChaDRG -> NewInvitation -> m InvitationId
+  createInvitation st gVar NewInvitation {contactConnId, connReq, recipientConnInfo} =
+    liftIOEither . withTransaction st $ \db ->
+      createWithRandomId gVar $ \invitationId ->
+        DB.execute
+          db
+          [sql|
+            INSERT INTO conn_invitations
+            (invitation_id,  contact_conn_id, cr_invitation, recipient_conn_info, accepted) VALUES (?, ?, ?, ?, 0);
+          |]
+          (invitationId, contactConnId, connReq, recipientConnInfo)
+
+  getInvitation :: SQLiteStore -> InvitationId -> m Invitation
+  getInvitation st invitationId =
+    liftIOEither . withTransaction st $ \db ->
+      firstRow invitation SEInvitationNotFound $
+        DB.query
+          db
+          [sql|
+            SELECT contact_conn_id, cr_invitation, recipient_conn_info, own_conn_info, accepted
+            FROM conn_invitations
+            WHERE invitation_id = ?
+              AND accepted = 0
+          |]
+          (Only invitationId)
+    where
+      invitation (contactConnId, connReq, recipientConnInfo, ownConnInfo, accepted) =
+        Invitation {invitationId, contactConnId, connReq, recipientConnInfo, ownConnInfo, accepted}
+
+  acceptInvitation :: SQLiteStore -> InvitationId -> ConnInfo -> m ()
+  acceptInvitation st invitationId ownConnInfo =
+    liftIO . withTransaction st $ \db -> do
+      DB.executeNamed
+        db
+        [sql|
+          UPDATE conn_invitations
+          SET accepted = 1,
+              own_conn_info = :own_conn_info
+          WHERE invitation_id = :invitation_id
+        |]
+        [ ":own_conn_info" := ownConnInfo,
+          ":invitation_id" := invitationId
+        ]
+
+  deleteInvitation :: SQLiteStore -> ConnId -> InvitationId -> m ()
+  deleteInvitation st contactConnId invId =
+    liftIOEither . withTransaction st $ \db ->
+      runExceptT $
+        ExceptT (getConn_ db contactConnId) >>= \case
+          SomeConn SCContact _ ->
+            liftIO $ DB.execute db "DELETE FROM conn_invitations WHERE contact_conn_id = ? AND invitation_id = ?" (contactConnId, invId)
+          _ -> throwError SEConnNotFound
+
+  updateRcvIds :: SQLiteStore -> ConnId -> m (InternalId, InternalRcvId, PrevExternalSndId, PrevRcvMsgHash)
+  updateRcvIds st connId =
+    liftIO . withTransaction st $ \db -> do
+      (lastInternalId, lastInternalRcvId, lastExternalSndId, lastRcvHash) <- retrieveLastIdsAndHashRcv_ db connId
+      let internalId = InternalId $ unId lastInternalId + 1
+          internalRcvId = InternalRcvId $ unRcvId lastInternalRcvId + 1
+      updateLastIdsRcv_ db connId internalId internalRcvId
+      pure (internalId, internalRcvId, lastExternalSndId, lastRcvHash)
+
+  createRcvMsg :: SQLiteStore -> ConnId -> RcvMsgData -> m ()
+  createRcvMsg st connId rcvMsgData =
+    liftIO . withTransaction st $ \db -> do
+      insertRcvMsgBase_ db connId rcvMsgData
+      insertRcvMsgDetails_ db connId rcvMsgData
+      updateHashRcv_ db connId rcvMsgData
+
+  updateSndIds :: SQLiteStore -> ConnId -> m (InternalId, InternalSndId, PrevSndMsgHash)
+  updateSndIds st connId =
+    liftIO . withTransaction st $ \db -> do
+      (lastInternalId, lastInternalSndId, prevSndHash) <- retrieveLastIdsAndHashSnd_ db connId
+      let internalId = InternalId $ unId lastInternalId + 1
+          internalSndId = InternalSndId $ unSndId lastInternalSndId + 1
+      updateLastIdsSnd_ db connId internalId internalSndId
+      pure (internalId, internalSndId, prevSndHash)
+
+  createSndMsg :: SQLiteStore -> ConnId -> SndMsgData -> m ()
+  createSndMsg st connId sndMsgData =
+    liftIO . withTransaction st $ \db -> do
+      insertSndMsgBase_ db connId sndMsgData
+      insertSndMsgDetails_ db connId sndMsgData
+      updateHashSnd_ db connId sndMsgData
+
+  getPendingMsgData :: SQLiteStore -> ConnId -> InternalId -> m (Maybe RcvQueue, (AMsgType, MsgBody, InternalTs))
+  getPendingMsgData st connId msgId =
+    liftIOEither . withTransaction st $ \db -> runExceptT $ do
+      rq_ <- liftIO $ getRcvQueueByConnId_ db connId
+      msgData <-
+        ExceptT . firstRow id SEMsgNotFound $
+          DB.query
+            db
+            [sql|
+                SELECT m.msg_type, m.msg_body, m.internal_ts
+                FROM messages m
+                JOIN snd_messages s ON s.conn_id = m.conn_id AND s.internal_id = m.internal_id
+                WHERE m.conn_id = ? AND m.internal_id = ?
+              |]
+            (connId, msgId)
+      pure (rq_, msgData)
+
+  getPendingMsgs :: SQLiteStore -> ConnId -> m [InternalId]
+  getPendingMsgs st connId =
+    liftIO . withTransaction st $ \db ->
+      map fromOnly
+        <$> DB.query db "SELECT internal_id FROM snd_messages WHERE conn_id = ?" (Only connId)
+
+  checkRcvMsg :: SQLiteStore -> ConnId -> InternalId -> m ()
+  checkRcvMsg st connId msgId =
+    liftIOEither . withTransaction st $ \db ->
+      hasMsg
+        <$> DB.query
+          db
+          [sql|
+            SELECT conn_id, internal_id
+            FROM rcv_messages
+            WHERE conn_id = ? AND internal_id = ?
+          |]
+          (connId, msgId)
+    where
+      hasMsg :: [(ConnId, InternalId)] -> Either StoreError ()
+      hasMsg r = if null r then Left SEMsgNotFound else Right ()
+
+  deleteMsg :: SQLiteStore -> ConnId -> InternalId -> m ()
+  deleteMsg st connId msgId =
+    liftIO . withTransaction st $ \db ->
+      DB.execute db "DELETE FROM messages WHERE conn_id = ? AND internal_id = ?;" (connId, msgId)
+
+  createRatchetX3dhKeys :: SQLiteStore -> ConnId -> C.PrivateKeyX448 -> C.PrivateKeyX448 -> m ()
+  createRatchetX3dhKeys st connId x3dhPrivKey1 x3dhPrivKey2 =
+    liftIO . withTransaction st $ \db ->
+      DB.execute db "INSERT INTO ratchets (conn_id, x3dh_priv_key_1, x3dh_priv_key_2) VALUES (?, ?, ?)" (connId, x3dhPrivKey1, x3dhPrivKey2)
+
+  getRatchetX3dhKeys :: SQLiteStore -> ConnId -> m (C.PrivateKeyX448, C.PrivateKeyX448)
+  getRatchetX3dhKeys st connId =
+    liftIOEither . withTransaction st $ \db ->
+      fmap hasKeys $
+        firstRow id SEX3dhKeysNotFound $
+          DB.query db "SELECT x3dh_priv_key_1, x3dh_priv_key_2 FROM ratchets WHERE conn_id = ?" (Only connId)
+    where
+      hasKeys = \case
+        Right (Just k1, Just k2) -> Right (k1, k2)
+        _ -> Left SEX3dhKeysNotFound
+
+  createRatchet :: SQLiteStore -> ConnId -> RatchetX448 -> m ()
+  createRatchet st connId rc =
+    liftIO . withTransaction st $ \db -> do
+      DB.executeNamed
+        db
+        [sql|
+          INSERT INTO ratchets (conn_id, ratchet_state)
+          VALUES (:conn_id, :ratchet_state)
+          ON CONFLICT (conn_id) DO UPDATE SET
+            ratchet_state = :ratchet_state,
+            x3dh_priv_key_1 = NULL,
+            x3dh_priv_key_2 = NULL
+        |]
+        [":conn_id" := connId, ":ratchet_state" := rc]
+
+  getRatchet :: SQLiteStore -> ConnId -> m RatchetX448
+  getRatchet st connId =
+    liftIOEither . withTransaction st $ \db ->
+      ratchet
+        <$> DB.query db "SELECT ratchet_state FROM ratchets WHERE conn_id = ?" (Only connId)
+    where
+      ratchet (Only (Just rc) : _) = Right rc
+      ratchet _ = Left SERatchetNotFound
+
+  getSkippedMsgKeys :: SQLiteStore -> ConnId -> m SkippedMsgKeys
+  getSkippedMsgKeys st connId =
+    liftIO . withTransaction st $ \db ->
+      skipped <$> DB.query db "SELECT header_key, msg_n, msg_key FROM skipped_messages WHERE conn_id = ?" (Only connId)
+    where
+      skipped ms = foldl' addSkippedKey M.empty ms
+      addSkippedKey smks (hk, msgN, mk) = M.alter (Just . addMsgKey) hk smks
+        where
+          addMsgKey = maybe (M.singleton msgN mk) (M.insert msgN mk)
+
+  updateRatchet :: SQLiteStore -> ConnId -> RatchetX448 -> SkippedMsgDiff -> m ()
+  updateRatchet st connId rc skipped =
+    liftIO . withTransaction st $ \db -> do
+      DB.execute db "UPDATE ratchets SET ratchet_state = ? WHERE conn_id = ?" (rc, connId)
+      case skipped of
+        SMDNoChange -> pure ()
+        SMDRemove hk msgN ->
+          DB.execute db "DELETE FROM skipped_messages WHERE conn_id = ? AND header_key = ? AND msg_n = ?" (connId, hk, msgN)
+        SMDAdd smks ->
+          forM_ (M.assocs smks) $ \(hk, mks) ->
+            forM_ (M.assocs mks) $ \(msgN, mk) ->
+              DB.execute db "INSERT INTO skipped_messages (conn_id, header_key, msg_n, msg_key) VALUES (?, ?, ?, ?)" (connId, hk, msgN, mk)
+
+-- * Auxiliary helpers
+
+instance ToField QueueStatus where toField = toField . serializeQueueStatus
+
+instance FromField QueueStatus where fromField = fromTextField_ queueStatusT
+
+instance ToField InternalRcvId where toField (InternalRcvId x) = toField x
+
+instance FromField InternalRcvId where fromField x = InternalRcvId <$> fromField x
+
+instance ToField InternalSndId where toField (InternalSndId x) = toField x
+
+instance FromField InternalSndId where fromField x = InternalSndId <$> fromField x
+
+instance ToField InternalId where toField (InternalId x) = toField x
+
+instance FromField InternalId where fromField x = InternalId <$> fromField x
+
+instance ToField AMsgType where toField = toField . smpEncode
+
+instance FromField AMsgType where fromField = blobFieldParser smpP
+
+instance ToField MsgIntegrity where toField = toField . strEncode
+
+instance FromField MsgIntegrity where fromField = blobFieldParser strP
+
+instance ToField SMPQueueUri where toField = toField . strEncode
+
+instance FromField SMPQueueUri where fromField = blobFieldParser strP
+
+instance ToField AConnectionRequestUri where toField = toField . strEncode
+
+instance FromField AConnectionRequestUri where fromField = blobFieldParser strP
+
+instance ConnectionModeI c => ToField (ConnectionRequestUri c) where toField = toField . strEncode
+
+instance (E.Typeable c, ConnectionModeI c) => FromField (ConnectionRequestUri c) where fromField = blobFieldParser strP
+
+instance ToField ConnectionMode where toField = toField . decodeLatin1 . strEncode
+
+instance FromField ConnectionMode where fromField = fromTextField_ connModeT
+
+instance ToField (SConnectionMode c) where toField = toField . connMode
+
+instance FromField AConnectionMode where fromField = fromTextField_ $ fmap connMode' . connModeT
+
+fromTextField_ :: (E.Typeable a) => (Text -> Maybe a) -> Field -> Ok a
+fromTextField_ fromText = \case
+  f@(Field (SQLText t) _) ->
+    case fromText t of
+      Just x -> Ok x
+      _ -> returnError ConversionFailed f ("invalid text: " <> T.unpack t)
+  f -> returnError ConversionFailed f "expecting SQLText column type"
+
+listToEither :: e -> [a] -> Either e a
+listToEither _ (x : _) = Right x
+listToEither e _ = Left e
+
+firstRow :: (a -> b) -> e -> IO [a] -> IO (Either e b)
+firstRow f e a = second f . listToEither e <$> a
+
+{- ORMOLU_DISABLE -}
+-- SQLite.Simple only has these up to 10 fields, which is insufficient for some of our queries
+instance (FromField a, FromField b, FromField c, FromField d, FromField e,
+          FromField f, FromField g, FromField h, FromField i, FromField j,
+          FromField k) =>
+  FromRow (a,b,c,d,e,f,g,h,i,j,k) where
+  fromRow = (,,,,,,,,,,) <$> field <*> field <*> field <*> field <*> field
+                         <*> field <*> field <*> field <*> field <*> field
+                         <*> field
+
+instance (FromField a, FromField b, FromField c, FromField d, FromField e,
+          FromField f, FromField g, FromField h, FromField i, FromField j,
+          FromField k, FromField l) =>
+  FromRow (a,b,c,d,e,f,g,h,i,j,k,l) where
+  fromRow = (,,,,,,,,,,,) <$> field <*> field <*> field <*> field <*> field
+                          <*> field <*> field <*> field <*> field <*> field
+                          <*> field <*> field
+
+instance (ToField a, ToField b, ToField c, ToField d, ToField e, ToField f,
+          ToField g, ToField h, ToField i, ToField j, ToField k, ToField l) =>
+  ToRow (a,b,c,d,e,f,g,h,i,j,k,l) where
+  toRow (a,b,c,d,e,f,g,h,i,j,k,l) =
+    [ toField a, toField b, toField c, toField d, toField e, toField f,
+      toField g, toField h, toField i, toField j, toField k, toField l
+    ]
+
+{- ORMOLU_ENABLE -}
+
+-- * Server upsert helper
+
+upsertServer_ :: DB.Connection -> SMPServer -> IO ()
+upsertServer_ dbConn SMPServer {host, port, keyHash} = do
+  DB.executeNamed
+    dbConn
+    [sql|
+      INSERT INTO servers (host, port, key_hash) VALUES (:host,:port,:key_hash)
+      ON CONFLICT (host, port) DO UPDATE SET
+        host=excluded.host,
+        port=excluded.port,
+        key_hash=excluded.key_hash;
+    |]
+    [":host" := host, ":port" := port, ":key_hash" := keyHash]
+
+-- * createRcvConn helpers
+
+insertRcvQueue_ :: DB.Connection -> ConnId -> RcvQueue -> IO ()
+insertRcvQueue_ dbConn connId RcvQueue {..} = do
+  DB.executeNamed
+    dbConn
+    [sql|
+      INSERT INTO rcv_queues
+        ( host, port, rcv_id, conn_id, rcv_private_key, rcv_dh_secret, e2e_priv_key, e2e_dh_secret, snd_id, status)
+      VALUES
+        (:host,:port,:rcv_id,:conn_id,:rcv_private_key,:rcv_dh_secret,:e2e_priv_key,:e2e_dh_secret,:snd_id,:status);
+    |]
+    [ ":host" := host server,
+      ":port" := port server,
+      ":rcv_id" := rcvId,
+      ":conn_id" := connId,
+      ":rcv_private_key" := rcvPrivateKey,
+      ":rcv_dh_secret" := rcvDhSecret,
+      ":e2e_priv_key" := e2ePrivKey,
+      ":e2e_dh_secret" := e2eDhSecret,
+      ":snd_id" := sndId,
+      ":status" := status
+    ]
+
+-- * createSndConn helpers
+
+insertSndQueue_ :: DB.Connection -> ConnId -> SndQueue -> IO ()
+insertSndQueue_ dbConn connId SndQueue {..} = do
+  DB.executeNamed
+    dbConn
+    [sql|
+      INSERT INTO snd_queues
+        ( host, port, snd_id, conn_id, snd_private_key, e2e_dh_secret, status)
+      VALUES
+        (:host,:port,:snd_id,:conn_id,:snd_private_key,:e2e_dh_secret,:status);
+    |]
+    [ ":host" := host server,
+      ":port" := port server,
+      ":snd_id" := sndId,
+      ":conn_id" := connId,
+      ":snd_private_key" := sndPrivateKey,
+      ":e2e_dh_secret" := e2eDhSecret,
+      ":status" := status
+    ]
+
+-- * getConn helpers
+
+getConn_ :: DB.Connection -> ConnId -> IO (Either StoreError SomeConn)
+getConn_ dbConn connId =
+  getConnData_ dbConn connId >>= \case
+    Nothing -> pure $ Left SEConnNotFound
+    Just (connData, cMode) -> do
+      rQ <- getRcvQueueByConnId_ dbConn connId
+      sQ <- getSndQueueByConnId_ dbConn connId
+      pure $ case (rQ, sQ, cMode) of
+        (Just rcvQ, Just sndQ, CMInvitation) -> Right $ SomeConn SCDuplex (DuplexConnection connData rcvQ sndQ)
+        (Just rcvQ, Nothing, CMInvitation) -> Right $ SomeConn SCRcv (RcvConnection connData rcvQ)
+        (Nothing, Just sndQ, CMInvitation) -> Right $ SomeConn SCSnd (SndConnection connData sndQ)
+        (Just rcvQ, Nothing, CMContact) -> Right $ SomeConn SCContact (ContactConnection connData rcvQ)
+        _ -> Left SEConnNotFound
+
+getConnData_ :: DB.Connection -> ConnId -> IO (Maybe (ConnData, ConnectionMode))
+getConnData_ dbConn connId' =
+  connData
+    <$> DB.query dbConn "SELECT conn_id, conn_mode FROM connections WHERE conn_id = ?;" (Only connId')
+  where
+    connData [(connId, cMode)] = Just (ConnData {connId}, cMode)
+    connData _ = Nothing
+
+getRcvQueueByConnId_ :: DB.Connection -> ConnId -> IO (Maybe RcvQueue)
+getRcvQueueByConnId_ dbConn connId =
+  rcvQueue
+    <$> DB.query
+      dbConn
+      [sql|
+        SELECT s.key_hash, q.host, q.port, q.rcv_id, q.rcv_private_key, q.rcv_dh_secret,
+          q.e2e_priv_key, q.e2e_dh_secret, q.snd_id, q.status
+        FROM rcv_queues q
+        INNER JOIN servers s ON q.host = s.host AND q.port = s.port
+        WHERE q.conn_id = ?;
+      |]
+      (Only connId)
+  where
+    rcvQueue [(keyHash, host, port, rcvId, rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret, sndId, status)] =
+      let server = SMPServer host port keyHash
+       in Just RcvQueue {server, rcvId, rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret, sndId, status}
+    rcvQueue _ = Nothing
+
+getSndQueueByConnId_ :: DB.Connection -> ConnId -> IO (Maybe SndQueue)
+getSndQueueByConnId_ dbConn connId =
+  sndQueue
+    <$> DB.query
+      dbConn
+      [sql|
+        SELECT s.key_hash, q.host, q.port, q.snd_id, q.snd_private_key, q.e2e_dh_secret, q.status
+        FROM snd_queues q
+        INNER JOIN servers s ON q.host = s.host AND q.port = s.port
+        WHERE q.conn_id = ?;
+      |]
+      (Only connId)
+  where
+    sndQueue [(keyHash, host, port, sndId, sndPrivateKey, e2eDhSecret, status)] =
+      let server = SMPServer host port keyHash
+       in Just SndQueue {server, sndId, sndPrivateKey, e2eDhSecret, status}
+    sndQueue _ = Nothing
+
+-- * updateRcvIds helpers
+
+retrieveLastIdsAndHashRcv_ :: DB.Connection -> ConnId -> IO (InternalId, InternalRcvId, PrevExternalSndId, PrevRcvMsgHash)
+retrieveLastIdsAndHashRcv_ dbConn connId = do
+  [(lastInternalId, lastInternalRcvId, lastExternalSndId, lastRcvHash)] <-
+    DB.queryNamed
+      dbConn
+      [sql|
+        SELECT last_internal_msg_id, last_internal_rcv_msg_id, last_external_snd_msg_id, last_rcv_msg_hash
+        FROM connections
+        WHERE conn_id = :conn_id;
+      |]
+      [":conn_id" := connId]
+  return (lastInternalId, lastInternalRcvId, lastExternalSndId, lastRcvHash)
+
+updateLastIdsRcv_ :: DB.Connection -> ConnId -> InternalId -> InternalRcvId -> IO ()
+updateLastIdsRcv_ dbConn connId newInternalId newInternalRcvId =
+  DB.executeNamed
+    dbConn
+    [sql|
+      UPDATE connections
+      SET last_internal_msg_id = :last_internal_msg_id,
+          last_internal_rcv_msg_id = :last_internal_rcv_msg_id
+      WHERE conn_id = :conn_id;
+    |]
+    [ ":last_internal_msg_id" := newInternalId,
+      ":last_internal_rcv_msg_id" := newInternalRcvId,
+      ":conn_id" := connId
+    ]
+
+-- * createRcvMsg helpers
+
+insertRcvMsgBase_ :: DB.Connection -> ConnId -> RcvMsgData -> IO ()
+insertRcvMsgBase_ dbConn connId RcvMsgData {msgMeta, msgType, msgBody, internalRcvId} = do
+  let MsgMeta {recipient = (internalId, internalTs)} = msgMeta
+  DB.executeNamed
+    dbConn
+    [sql|
+      INSERT INTO messages
+        ( conn_id, internal_id, internal_ts, internal_rcv_id, internal_snd_id, msg_type, msg_body)
+      VALUES
+        (:conn_id,:internal_id,:internal_ts,:internal_rcv_id,            NULL,:msg_type, :msg_body);
+    |]
+    [ ":conn_id" := connId,
+      ":internal_id" := internalId,
+      ":internal_ts" := internalTs,
+      ":internal_rcv_id" := internalRcvId,
+      ":msg_type" := msgType,
+      ":msg_body" := msgBody
+    ]
+
+insertRcvMsgDetails_ :: DB.Connection -> ConnId -> RcvMsgData -> IO ()
+insertRcvMsgDetails_ dbConn connId RcvMsgData {msgMeta, internalRcvId, internalHash, externalPrevSndHash} = do
+  let MsgMeta {integrity, recipient, broker, sndMsgId} = msgMeta
+  DB.executeNamed
+    dbConn
+    [sql|
+      INSERT INTO rcv_messages
+        ( conn_id, internal_rcv_id, internal_id, external_snd_id,
+          broker_id, broker_ts,
+          internal_hash, external_prev_snd_hash, integrity)
+      VALUES
+        (:conn_id,:internal_rcv_id,:internal_id,:external_snd_id,
+         :broker_id,:broker_ts,
+         :internal_hash,:external_prev_snd_hash,:integrity);
+    |]
+    [ ":conn_id" := connId,
+      ":internal_rcv_id" := internalRcvId,
+      ":internal_id" := fst recipient,
+      ":external_snd_id" := sndMsgId,
+      ":broker_id" := fst broker,
+      ":broker_ts" := snd broker,
+      ":internal_hash" := internalHash,
+      ":external_prev_snd_hash" := externalPrevSndHash,
+      ":integrity" := integrity
+    ]
+
+updateHashRcv_ :: DB.Connection -> ConnId -> RcvMsgData -> IO ()
+updateHashRcv_ dbConn connId RcvMsgData {msgMeta, internalHash, internalRcvId} =
+  DB.executeNamed
+    dbConn
+    -- last_internal_rcv_msg_id equality check prevents race condition in case next id was reserved
+    [sql|
+      UPDATE connections
+      SET last_external_snd_msg_id = :last_external_snd_msg_id,
+          last_rcv_msg_hash = :last_rcv_msg_hash
+      WHERE conn_id = :conn_id
+        AND last_internal_rcv_msg_id = :last_internal_rcv_msg_id;
+    |]
+    [ ":last_external_snd_msg_id" := sndMsgId (msgMeta :: MsgMeta),
+      ":last_rcv_msg_hash" := internalHash,
+      ":conn_id" := connId,
+      ":last_internal_rcv_msg_id" := internalRcvId
+    ]
+
+-- * updateSndIds helpers
+
+retrieveLastIdsAndHashSnd_ :: DB.Connection -> ConnId -> IO (InternalId, InternalSndId, PrevSndMsgHash)
+retrieveLastIdsAndHashSnd_ dbConn connId = do
+  [(lastInternalId, lastInternalSndId, lastSndHash)] <-
+    DB.queryNamed
+      dbConn
+      [sql|
+        SELECT last_internal_msg_id, last_internal_snd_msg_id, last_snd_msg_hash
+        FROM connections
+        WHERE conn_id = :conn_id;
+      |]
+      [":conn_id" := connId]
+  return (lastInternalId, lastInternalSndId, lastSndHash)
+
+updateLastIdsSnd_ :: DB.Connection -> ConnId -> InternalId -> InternalSndId -> IO ()
+updateLastIdsSnd_ dbConn connId newInternalId newInternalSndId =
+  DB.executeNamed
+    dbConn
+    [sql|
+      UPDATE connections
+      SET last_internal_msg_id = :last_internal_msg_id,
+          last_internal_snd_msg_id = :last_internal_snd_msg_id
+      WHERE conn_id = :conn_id;
+    |]
+    [ ":last_internal_msg_id" := newInternalId,
+      ":last_internal_snd_msg_id" := newInternalSndId,
+      ":conn_id" := connId
+    ]
+
+-- * createSndMsg helpers
+
+insertSndMsgBase_ :: DB.Connection -> ConnId -> SndMsgData -> IO ()
+insertSndMsgBase_ dbConn connId SndMsgData {..} = do
+  DB.executeNamed
+    dbConn
+    [sql|
+      INSERT INTO messages
+        ( conn_id, internal_id, internal_ts, internal_rcv_id, internal_snd_id, msg_type, msg_body)
+      VALUES
+        (:conn_id,:internal_id,:internal_ts,            NULL,:internal_snd_id,:msg_type, :msg_body);
+    |]
+    [ ":conn_id" := connId,
+      ":internal_id" := internalId,
+      ":internal_ts" := internalTs,
+      ":internal_snd_id" := internalSndId,
+      ":msg_type" := msgType,
+      ":msg_body" := msgBody
+    ]
+
+insertSndMsgDetails_ :: DB.Connection -> ConnId -> SndMsgData -> IO ()
+insertSndMsgDetails_ dbConn connId SndMsgData {..} =
+  DB.executeNamed
+    dbConn
+    [sql|
+      INSERT INTO snd_messages
+        ( conn_id, internal_snd_id, internal_id, internal_hash, previous_msg_hash)
+      VALUES
+        (:conn_id,:internal_snd_id,:internal_id,:internal_hash,:previous_msg_hash);
+    |]
+    [ ":conn_id" := connId,
+      ":internal_snd_id" := internalSndId,
+      ":internal_id" := internalId,
+      ":internal_hash" := internalHash,
+      ":previous_msg_hash" := prevMsgHash
+    ]
+
+updateHashSnd_ :: DB.Connection -> ConnId -> SndMsgData -> IO ()
+updateHashSnd_ dbConn connId SndMsgData {..} =
+  DB.executeNamed
+    dbConn
+    -- last_internal_snd_msg_id equality check prevents race condition in case next id was reserved
+    [sql|
+      UPDATE connections
+      SET last_snd_msg_hash = :last_snd_msg_hash
+      WHERE conn_id = :conn_id
+        AND last_internal_snd_msg_id = :last_internal_snd_msg_id;
+    |]
+    [ ":last_snd_msg_hash" := internalHash,
+      ":conn_id" := connId,
+      ":last_internal_snd_msg_id" := internalSndId
+    ]
+
+-- create record with a random ID
+createWithRandomId :: TVar ChaChaDRG -> (ByteString -> IO ()) -> IO (Either StoreError ByteString)
+createWithRandomId gVar create = tryCreate 3
+  where
+    tryCreate :: Int -> IO (Either StoreError ByteString)
+    tryCreate 0 = pure $ Left SEUniqueID
+    tryCreate n = do
+      id' <- randomId gVar 12
+      E.try (create id') >>= \case
+        Right _ -> pure $ Right id'
+        Left e
+          | DB.sqlError e == DB.ErrorConstraint -> tryCreate (n - 1)
+          | otherwise -> pure . Left . SEInternal $ bshow e
+
+randomId :: TVar ChaChaDRG -> Int -> IO ByteString
+randomId gVar n = U.encode <$> (atomically . stateTVar gVar $ randomBytesGenerate n)

src/Simplex/Messaging/Agent/Store/Postgres/Migrations.hs

@@ -0,0 +1,72 @@
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE QuasiQuotes #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TemplateHaskell #-}
+{-# LANGUAGE TupleSections #-}
+
+module Simplex.Messaging.Agent.Store.Postgres.Migrations
+  ( Migration (..),
+    app,
+    initialize,
+    get,
+    run,
+  )
+where
+
+import Control.Monad (forM_)
+import Data.Function (on)
+import Data.List (intercalate, sortBy)
+import Data.Text (Text)
+import Data.Time.Clock (getCurrentTime)
+import Database.SQLite.Simple (Connection, Only (..), Query (..))
+import qualified Database.SQLite.Simple as DB
+import Database.SQLite.Simple.QQ (sql)
+import qualified Database.SQLite3 as SQLite3
+import Simplex.Messaging.Agent.Store.Postgres.Migrations.M20220202_initial
+
+data Migration = Migration {name :: String, up :: Text}
+  deriving (Show)
+
+schemaMigrations :: [(String, Query)]
+schemaMigrations =
+  [ ("20220101_initial", m20220202_initial)
+  ]
+
+-- | The list of migrations in ascending order by date
+app :: [Migration]
+app = sortBy (compare `on` name) $ map migration schemaMigrations
+  where
+    migration (name, query) = Migration {name = name, up = fromQuery query}
+
+get :: Connection -> [Migration] -> IO (Either String [Migration])
+get conn migrations =
+  migrationsToRun migrations . map fromOnly
+    <$> DB.query_ conn "SELECT name FROM migrations ORDER BY name ASC;"
+
+run :: Connection -> [Migration] -> IO ()
+run conn ms = DB.withImmediateTransaction conn . forM_ ms $
+  \Migration {name, up} -> insert name >> execSQL up
+  where
+    insert name = DB.execute conn "INSERT INTO migrations (name, ts) VALUES (?, ?);" . (name,) =<< getCurrentTime
+    execSQL = SQLite3.exec $ DB.connectionHandle conn
+
+initialize :: Connection -> IO ()
+initialize conn =
+  DB.execute_
+    conn
+    [sql|
+      CREATE TABLE IF NOT EXISTS migrations (
+        name TEXT NOT NULL,
+        ts TEXT NOT NULL,
+        PRIMARY KEY (name)
+      );
+    |]
+
+migrationsToRun :: [Migration] -> [String] -> Either String [Migration]
+migrationsToRun appMs [] = Right appMs
+migrationsToRun [] dbMs = Left $ "database version is newer than the app: " <> intercalate ", " dbMs
+migrationsToRun (a : as) (d : ds)
+  | name a == d = migrationsToRun as ds
+  | otherwise = Left $ "different migration in the app/database: " <> name a <> " / " <> d

src/Simplex/Messaging/Agent/Store/Postgres/Migrations/M20220202_initial.hs

@@ -0,0 +1,148 @@
+{-# LANGUAGE QuasiQuotes #-}
+
+module Simplex.Messaging.Agent.Store.Postgres.Migrations.M20220202_initial where
+
+import Database.SQLite.Simple (Query)
+import Database.SQLite.Simple.QQ (sql)
+
+m20220202_initial :: Query
+m20220202_initial =
+  [sql|
+CREATE TABLE servers (
+  host TEXT NOT NULL,
+  port TEXT NOT NULL,
+  key_hash BLOB NOT NULL,
+  PRIMARY KEY (host, port)
+) WITHOUT ROWID;
+
+CREATE TABLE connections (
+  conn_id BLOB NOT NULL PRIMARY KEY,
+  conn_mode TEXT NOT NULL,
+  last_internal_msg_id INTEGER NOT NULL DEFAULT 0,
+  last_internal_rcv_msg_id INTEGER NOT NULL DEFAULT 0,
+  last_internal_snd_msg_id INTEGER NOT NULL DEFAULT 0,
+  last_external_snd_msg_id INTEGER NOT NULL DEFAULT 0,
+  last_rcv_msg_hash BLOB NOT NULL DEFAULT x'',
+  last_snd_msg_hash BLOB NOT NULL DEFAULT x'',
+  smp_agent_version INTEGER NOT NULL DEFAULT 1
+) WITHOUT ROWID;
+
+CREATE TABLE rcv_queues (
+  host TEXT NOT NULL,
+  port TEXT NOT NULL,
+  rcv_id BLOB NOT NULL,
+  conn_id BLOB NOT NULL REFERENCES connections ON DELETE CASCADE,
+  rcv_private_key BLOB NOT NULL,
+  rcv_dh_secret BLOB NOT NULL,
+  e2e_priv_key BLOB NOT NULL,
+  e2e_dh_secret BLOB,
+  snd_id BLOB NOT NULL,
+  snd_key BLOB,
+  status TEXT NOT NULL,
+  smp_server_version INTEGER NOT NULL DEFAULT 1,
+  smp_client_version INTEGER,
+  PRIMARY KEY (host, port, rcv_id),
+  FOREIGN KEY (host, port) REFERENCES servers
+    ON DELETE RESTRICT ON UPDATE CASCADE,
+  UNIQUE (host, port, snd_id)
+) WITHOUT ROWID;
+
+CREATE TABLE snd_queues (
+  host TEXT NOT NULL,
+  port TEXT NOT NULL,
+  snd_id BLOB NOT NULL,
+  conn_id BLOB NOT NULL REFERENCES connections ON DELETE CASCADE,
+  snd_private_key BLOB NOT NULL,
+  e2e_dh_secret BLOB NOT NULL,
+  status TEXT NOT NULL,
+  smp_server_version INTEGER NOT NULL DEFAULT 1,
+  smp_client_version INTEGER NOT NULL DEFAULT 1,
+  PRIMARY KEY (host, port, snd_id),
+  FOREIGN KEY (host, port) REFERENCES servers
+    ON DELETE RESTRICT ON UPDATE CASCADE
+) WITHOUT ROWID;
+
+CREATE TABLE messages (
+  conn_id BLOB NOT NULL REFERENCES connections (conn_id)
+    ON DELETE CASCADE,
+  internal_id INTEGER NOT NULL,
+  internal_ts TEXT NOT NULL,
+  internal_rcv_id INTEGER,
+  internal_snd_id INTEGER,
+  msg_type BLOB NOT NULL, -- (H)ELLO, (R)EPLY, (D)ELETE. Should SMP confirmation be saved too?
+  msg_body BLOB NOT NULL DEFAULT x'',
+  PRIMARY KEY (conn_id, internal_id),
+  FOREIGN KEY (conn_id, internal_rcv_id) REFERENCES rcv_messages
+    ON DELETE CASCADE DEFERRABLE INITIALLY DEFERRED,
+  FOREIGN KEY (conn_id, internal_snd_id) REFERENCES snd_messages
+    ON DELETE CASCADE DEFERRABLE INITIALLY DEFERRED
+) WITHOUT ROWID;
+
+CREATE TABLE rcv_messages (
+  conn_id BLOB NOT NULL,
+  internal_rcv_id INTEGER NOT NULL,
+  internal_id INTEGER NOT NULL,
+  external_snd_id INTEGER NOT NULL,
+  broker_id BLOB NOT NULL,
+  broker_ts TEXT NOT NULL,
+  internal_hash BLOB NOT NULL,
+  external_prev_snd_hash BLOB NOT NULL,
+  integrity BLOB NOT NULL,
+  PRIMARY KEY (conn_id, internal_rcv_id),
+  FOREIGN KEY (conn_id, internal_id) REFERENCES messages
+    ON DELETE CASCADE
+) WITHOUT ROWID;
+
+CREATE TABLE snd_messages (
+  conn_id BLOB NOT NULL,
+  internal_snd_id INTEGER NOT NULL,
+  internal_id INTEGER NOT NULL,
+  internal_hash BLOB NOT NULL,
+  previous_msg_hash BLOB NOT NULL DEFAULT x'',
+  PRIMARY KEY (conn_id, internal_snd_id),
+  FOREIGN KEY (conn_id, internal_id) REFERENCES messages
+    ON DELETE CASCADE
+) WITHOUT ROWID;
+
+CREATE TABLE conn_confirmations (
+  confirmation_id BLOB NOT NULL PRIMARY KEY,
+  conn_id BLOB NOT NULL REFERENCES connections ON DELETE CASCADE,
+  e2e_snd_pub_key BLOB NOT NULL, -- TODO per-queue key. Split?
+  sender_key BLOB NOT NULL, -- TODO per-queue key. Split?
+  ratchet_state BLOB NOT NULL,
+  sender_conn_info BLOB NOT NULL,
+  accepted INTEGER NOT NULL,
+  own_conn_info BLOB,
+  created_at TEXT NOT NULL DEFAULT (datetime('now'))
+) WITHOUT ROWID;
+
+CREATE TABLE conn_invitations (
+  invitation_id BLOB NOT NULL PRIMARY KEY,
+  contact_conn_id BLOB NOT NULL REFERENCES connections ON DELETE CASCADE,
+  cr_invitation BLOB NOT NULL,
+  recipient_conn_info BLOB NOT NULL,
+  accepted INTEGER NOT NULL DEFAULT 0,
+  own_conn_info BLOB,
+  created_at TEXT NOT NULL DEFAULT (datetime('now'))
+) WITHOUT ROWID;
+
+CREATE TABLE ratchets (
+  conn_id BLOB NOT NULL PRIMARY KEY REFERENCES connections
+    ON DELETE CASCADE,
+  -- x3dh keys are not saved on the sending side (the side accepting the connection)
+  x3dh_priv_key_1 BLOB,
+  x3dh_priv_key_2 BLOB,
+  -- ratchet is initially empty on the receiving side (the side offering the connection)
+  ratchet_state BLOB,
+  e2e_version INTEGER NOT NULL DEFAULT 1
+) WITHOUT ROWID;
+
+CREATE TABLE skipped_messages (
+  skipped_message_id INTEGER PRIMARY KEY,
+  conn_id BLOB NOT NULL REFERENCES ratchets
+    ON DELETE CASCADE,
+  header_key BLOB NOT NULL,
+  msg_n INTEGER NOT NULL,
+  msg_key BLOB NOT NULL
+);
+|]

src/Simplex/Messaging/Agent/Store/SQLite.hs

@@ -440,7 +440,7 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto
       insertSndMsgDetails_ db connId sndMsgData
       updateHashSnd_ db connId sndMsgData
 
-  getPendingMsgData :: SQLiteStore -> ConnId -> InternalId -> m (Maybe RcvQueue, (AMsgType, MsgBody))
+  getPendingMsgData :: SQLiteStore -> ConnId -> InternalId -> m (Maybe RcvQueue, (AMsgType, MsgBody, InternalTs))
   getPendingMsgData st connId msgId =
     liftIOEither . withTransaction st $ \db -> runExceptT $ do
       rq_ <- liftIO $ getRcvQueueByConnId_ db connId
@@ -449,7 +449,7 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto
           DB.query
             db
             [sql|
-                SELECT m.msg_type, m.msg_body
+                SELECT m.msg_type, m.msg_body, m.internal_ts
                 FROM messages m
                 JOIN snd_messages s ON s.conn_id = m.conn_id AND s.internal_id = m.internal_id
                 WHERE m.conn_id = ? AND m.internal_id = ?
@@ -569,9 +569,9 @@ instance ToField AMsgType where toField = toField . smpEncode
 
 instance FromField AMsgType where fromField = blobFieldParser smpP
 
-instance ToField MsgIntegrity where toField = toField . serializeMsgIntegrity
+instance ToField MsgIntegrity where toField = toField . strEncode
 
-instance FromField MsgIntegrity where fromField = blobFieldParser msgIntegrityP
+instance FromField MsgIntegrity where fromField = blobFieldParser strP
 
 instance ToField SMPQueueUri where toField = toField . strEncode
 

src/Simplex/Messaging/Agent/Store/SQLite/Migrations.hs

@@ -16,29 +16,29 @@ module Simplex.Messaging.Agent.Store.SQLite.Migrations
 where
 
 import Control.Monad (forM_)
-import Data.FileEmbed (embedDir, makeRelativeToProject)
 import Data.Function (on)
 import Data.List (intercalate, sortBy)
 import Data.Text (Text)
-import Data.Text.Encoding (decodeUtf8)
 import Data.Time.Clock (getCurrentTime)
-import Database.SQLite.Simple (Connection, Only (..))
+import Database.SQLite.Simple (Connection, Only (..), Query (..))
 import qualified Database.SQLite.Simple as DB
 import Database.SQLite.Simple.QQ (sql)
 import qualified Database.SQLite3 as SQLite3
-import System.FilePath (takeBaseName, takeExtension)
+import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220101_initial
 
 data Migration = Migration {name :: String, up :: Text}
   deriving (Show)
 
+schemaMigrations :: [(String, Query)]
+schemaMigrations =
+  [ ("20220101_initial", m20220101_initial)
+  ]
+
 -- | The list of migrations in ascending order by date
 app :: [Migration]
-app =
-  sortBy (compare `on` name) . map migration . filter sqlFile $
-    $(makeRelativeToProject "migrations" >>= embedDir)
+app = sortBy (compare `on` name) $ map migration schemaMigrations
   where
-    sqlFile (file, _) = takeExtension file == ".sql"
-    migration (file, qStr) = Migration {name = takeBaseName file, up = decodeUtf8 qStr}
+    migration (name, query) = Migration {name = name, up = fromQuery query}
 
 get :: Connection -> [Migration] -> IO (Either String [Migration])
 get conn migrations =

src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20220101_initial.hs

@@ -1,3 +1,13 @@
+{-# LANGUAGE QuasiQuotes #-}
+
+module Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220101_initial where
+
+import Database.SQLite.Simple (Query)
+import Database.SQLite.Simple.QQ (sql)
+
+m20220101_initial :: Query
+m20220101_initial =
+  [sql|
 CREATE TABLE servers (
   host TEXT NOT NULL,
   port TEXT NOT NULL,
@@ -135,3 +145,4 @@ CREATE TABLE skipped_messages (
   msg_n INTEGER NOT NULL,
   msg_key BLOB NOT NULL
 );
+|]

src/Simplex/Messaging/Client.hs

@@ -63,7 +63,8 @@ import Network.Socket (ServiceName)
 import Numeric.Natural
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Protocol
-import Simplex.Messaging.Transport (ATransport (..), THandle (..), TLS, TProxy, Transport (..), TransportError, clientHandshake, runTransportClient)
+import Simplex.Messaging.Transport (ATransport (..), THandle (..), TLS, TProxy, Transport (..), TransportError, clientHandshake)
+import Simplex.Messaging.Transport.Client (runTransportClient)
 import Simplex.Messaging.Transport.WebSockets (WS)
 import Simplex.Messaging.Util (bshow, liftError, raceAny_)
 import System.Timeout (timeout)

src/Simplex/Messaging/Parsers.hs

@@ -4,13 +4,14 @@
 module Simplex.Messaging.Parsers where
 
 import Control.Monad.Trans.Except
+import qualified Data.Aeson as J
 import Data.Attoparsec.ByteString.Char8 (Parser)
 import qualified Data.Attoparsec.ByteString.Char8 as A
 import Data.Bifunctor (first)
 import Data.ByteString.Base64
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
-import Data.Char (isAlphaNum)
+import Data.Char (isAlphaNum, toLower)
 import Data.Time.Clock (UTCTime)
 import Data.Time.ISO8601 (parseISO8601)
 import Data.Typeable (Typeable)
@@ -78,3 +79,40 @@ blobFieldDecoder dec = \case
       Right k -> Ok k
       Left e -> returnError ConversionFailed f ("couldn't parse field: " ++ e)
   f -> returnError ConversionFailed f "expecting SQLBlob column type"
+
+fstToLower :: String -> String
+fstToLower "" = ""
+fstToLower (h : t) = toLower h : t
+
+dropPrefix :: String -> String -> String
+dropPrefix pfx s =
+  let (p, rest) = splitAt (length pfx) s
+   in fstToLower $ if p == pfx then rest else s
+
+enumJSON :: (String -> String) -> J.Options
+enumJSON tagModifier =
+  J.defaultOptions
+    { J.constructorTagModifier = tagModifier,
+      J.allNullaryToStringTag = True
+    }
+
+sumTypeJSON :: (String -> String) -> J.Options
+sumTypeJSON = singleFieldJSON
+
+taggedObjectJSON :: (String -> String) -> J.Options
+taggedObjectJSON tagModifier =
+  J.defaultOptions
+    { J.sumEncoding = J.TaggedObject "type" "data",
+      J.constructorTagModifier = tagModifier,
+      J.nullaryToObject = True,
+      J.omitNothingFields = True
+    }
+
+singleFieldJSON :: (String -> String) -> J.Options
+singleFieldJSON tagModifier =
+  J.defaultOptions
+    { J.sumEncoding = J.ObjectWithSingleField,
+      J.constructorTagModifier = tagModifier,
+      J.nullaryToObject = True,
+      J.omitNothingFields = True
+    }

src/Simplex/Messaging/Protocol.hs

@@ -90,6 +90,8 @@ where
 
 import Control.Applicative (optional, (<|>))
 import Control.Monad.Except
+import Data.Aeson (ToJSON (..))
+import qualified Data.Aeson as J
 import Data.Attoparsec.ByteString.Char8 (Parser)
 import qualified Data.Attoparsec.ByteString.Char8 as A
 import Data.ByteString.Char8 (ByteString)
@@ -107,7 +109,7 @@ import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers
 import Simplex.Messaging.Transport (THandle (..), Transport, TransportError (..), tGetBlock, tPutBlock)
-import Simplex.Messaging.Util ((<$?>))
+import Simplex.Messaging.Util (bshow, (<$?>))
 import Simplex.Messaging.Version
 import Test.QuickCheck (Arbitrary (..))
 
@@ -413,6 +415,15 @@ newtype CorrId = CorrId {bs :: ByteString} deriving (Eq, Ord, Show)
 instance IsString CorrId where
   fromString = CorrId . fromString
 
+instance StrEncoding CorrId where
+  strEncode (CorrId cId) = strEncode cId
+  strDecode s = CorrId <$> strDecode s
+  strP = CorrId <$> strP
+
+instance ToJSON CorrId where
+  toJSON = strToJSON
+  toEncoding = strToJEncoding
+
 -- | Queue IDs and keys
 data QueueIdsKeys = QIK
   { rcvId :: RecipientId,
@@ -462,7 +473,7 @@ data ErrorType
   | -- | incorrect SMP session ID (TLS Finished message / tls-unique binding RFC5929)
     SESSION
   | -- | SMP command is unknown or has invalid syntax
-    CMD CommandError
+    CMD {cmdErr :: CommandError}
   | -- | command authorization error - bad signature or non-existing SMP queue
     AUTH
   | -- | SMP queue capacity is exceeded on the server
@@ -477,6 +488,16 @@ data ErrorType
     DUPLICATE_ -- TODO remove, not part of SMP protocol
   deriving (Eq, Generic, Read, Show)
 
+instance ToJSON ErrorType where
+  toJSON = J.genericToJSON $ sumTypeJSON id
+  toEncoding = J.genericToEncoding $ sumTypeJSON id
+
+instance StrEncoding ErrorType where
+  strEncode = \case
+    CMD e -> "CMD " <> bshow e
+    e -> bshow e
+  strP = "CMD " *> (CMD <$> parseRead1) <|> parseRead1
+
 -- | SMP command error type.
 data CommandError
   = -- | unknown command
@@ -491,6 +512,10 @@ data CommandError
     NO_QUEUE
   deriving (Eq, Generic, Read, Show)
 
+instance ToJSON CommandError where
+  toJSON = J.genericToJSON $ sumTypeJSON id
+  toEncoding = J.genericToEncoding $ sumTypeJSON id
+
 instance Arbitrary ErrorType where arbitrary = genericArbitraryU
 
 instance Arbitrary CommandError where arbitrary = genericArbitraryU

src/Simplex/Messaging/Server.hs

@@ -48,6 +48,7 @@ import Simplex.Messaging.Server.QueueStore
 import Simplex.Messaging.Server.QueueStore.STM (QueueStore)
 import Simplex.Messaging.Server.StoreLog
 import Simplex.Messaging.Transport
+import Simplex.Messaging.Transport.Server
 import Simplex.Messaging.Util
 import UnliftIO.Concurrent
 import UnliftIO.Exception

src/Simplex/Messaging/Server/Env/STM.hs

@@ -21,7 +21,8 @@ import Simplex.Messaging.Server.MsgStore.STM
 import Simplex.Messaging.Server.QueueStore (QueueRec (..))
 import Simplex.Messaging.Server.QueueStore.STM
 import Simplex.Messaging.Server.StoreLog
-import Simplex.Messaging.Transport (ATransport, loadFingerprint, loadTLSServerParams)
+import Simplex.Messaging.Transport (ATransport)
+import Simplex.Messaging.Transport.Server (loadFingerprint, loadTLSServerParams)
 import System.IO (IOMode (..))
 import UnliftIO.STM
 

src/Simplex/Messaging/Transport.hs

@@ -36,15 +36,11 @@ module Simplex.Messaging.Transport
     ATransport (..),
     TransportPeer (..),
 
-    -- * Transport over TLS 1.2
-    runTransportServer,
-    runTransportClient,
-    loadTLSServerParams,
-    loadFingerprint,
-
-    -- * TLS 1.2 Transport
+    -- * TLS Transport
     TLS (..),
+    connectTLS,
     closeTLS,
+    supportedParameters,
     withTlsUnique,
 
     -- * SMP transport
@@ -64,9 +60,9 @@ where
 
 import Control.Applicative ((<|>))
 import Control.Monad.Except
-import Control.Monad.IO.Unlift
 import Control.Monad.Trans.Except (throwE)
-import qualified Crypto.Store.X509 as SX
+import Data.Aeson (ToJSON)
+import qualified Data.Aeson as J
 import Data.Attoparsec.ByteString.Char8 (Parser)
 import Data.Bifunctor (first)
 import Data.Bitraversable (bimapM)
@@ -75,14 +71,7 @@ import qualified Data.ByteString.Char8 as B
 import qualified Data.ByteString.Lazy as BL
 import Data.Default (def)
 import Data.Functor (($>))
-import Data.Set (Set)
-import qualified Data.Set as S
-import qualified Data.X509 as X
-import qualified Data.X509.CertificateStore as XS
-import Data.X509.Validation (Fingerprint (..))
-import qualified Data.X509.Validation as XV
 import GHC.Generics (Generic)
-import GHC.IO.Exception (IOErrorType (..))
 import GHC.IO.Handle.Internals (ioe_EOF)
 import Generic.Random (genericArbitraryU)
 import Network.Socket
@@ -90,14 +79,11 @@ import qualified Network.TLS as T
 import qualified Network.TLS.Extra as TE
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
-import Simplex.Messaging.Parsers (parse, parseRead1)
+import Simplex.Messaging.Parsers (dropPrefix, parse, parseRead1, sumTypeJSON)
 import Simplex.Messaging.Util (bshow)
 import Simplex.Messaging.Version
-import System.Exit (exitFailure)
-import System.IO.Error
 import Test.QuickCheck (Arbitrary (..))
-import UnliftIO.Concurrent
-import UnliftIO.Exception (Exception, IOException)
+import UnliftIO.Exception (Exception)
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
 
@@ -110,7 +96,7 @@ supportedSMPVersions :: VersionRange
 supportedSMPVersions = mkVersionRange 1 1
 
 simplexMQVersion :: String
-simplexMQVersion = "1.0.1"
+simplexMQVersion = "1.0.2"
 
 -- * Transport connection class
 
@@ -154,104 +140,7 @@ data TProxy c = TProxy
 
 data ATransport = forall c. Transport c => ATransport (TProxy c)
 
--- * Transport over TLS 1.2
-
--- | Run transport server (plain TCP or WebSockets) on passed TCP port and signal when server started and stopped via passed TMVar.
---
--- All accepted connections are passed to the passed function.
-runTransportServer :: forall c m. (Transport c, MonadUnliftIO m) => TMVar Bool -> ServiceName -> T.ServerParams -> (c -> m ()) -> m ()
-runTransportServer started port serverParams server = do
-  clients <- newTVarIO S.empty
-  E.bracket
-    (liftIO $ startTCPServer started port)
-    (liftIO . closeServer clients)
-    $ \sock -> forever $ connectClients sock clients `E.catch` \(_ :: E.SomeException) -> pure ()
-  where
-    connectClients :: Socket -> TVar (Set ThreadId) -> m ()
-    connectClients sock clients = do
-      c <- liftIO $ acceptConnection sock
-      tid <- server c `forkFinally` const (liftIO $ closeConnection c)
-      atomically . modifyTVar clients $ S.insert tid
-    closeServer :: TVar (Set ThreadId) -> Socket -> IO ()
-    closeServer clients sock = do
-      readTVarIO clients >>= mapM_ killThread
-      close sock
-      void . atomically $ tryPutTMVar started False
-    acceptConnection :: Socket -> IO c
-    acceptConnection sock = do
-      (newSock, _) <- accept sock
-      ctx <- connectTLS serverParams newSock
-      getServerConnection ctx
-
-startTCPServer :: TMVar Bool -> ServiceName -> IO Socket
-startTCPServer started port = withSocketsDo $ resolve >>= open >>= setStarted
-  where
-    resolve =
-      let hints = defaultHints {addrFlags = [AI_PASSIVE], addrSocketType = Stream}
-       in head <$> getAddrInfo (Just hints) Nothing (Just port)
-    open addr = do
-      sock <- socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr)
-      setSocketOption sock ReuseAddr 1
-      withFdSocket sock setCloseOnExecIfNeeded
-      bind sock $ addrAddress addr
-      listen sock 1024
-      return sock
-    setStarted sock = atomically (tryPutTMVar started True) >> pure sock
-
--- | Connect to passed TCP host:port and pass handle to the client.
-runTransportClient :: Transport c => MonadUnliftIO m => HostName -> ServiceName -> C.KeyHash -> (c -> m a) -> m a
-runTransportClient host port keyHash client = do
-  let clientParams = mkTLSClientParams host port keyHash
-  c <- liftIO $ startTCPClient host port clientParams
-  client c `E.finally` liftIO (closeConnection c)
-
-startTCPClient :: forall c. Transport c => HostName -> ServiceName -> T.ClientParams -> IO c
-startTCPClient host port clientParams = withSocketsDo $ resolve >>= tryOpen err
-  where
-    err :: IOException
-    err = mkIOError NoSuchThing "no address" Nothing Nothing
-
-    resolve :: IO [AddrInfo]
-    resolve =
-      let hints = defaultHints {addrSocketType = Stream}
-       in getAddrInfo (Just hints) (Just host) (Just port)
-
-    tryOpen :: IOException -> [AddrInfo] -> IO c
-    tryOpen e [] = E.throwIO e
-    tryOpen _ (addr : as) =
-      E.try (open addr) >>= either (`tryOpen` as) pure
-
-    open :: AddrInfo -> IO c
-    open addr = do
-      sock <- socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr)
-      connect sock $ addrAddress addr
-      ctx <- connectTLS clientParams sock
-      getClientConnection ctx
-
-loadTLSServerParams :: FilePath -> FilePath -> FilePath -> IO T.ServerParams
-loadTLSServerParams caCertificateFile certificateFile privateKeyFile =
-  fromCredential <$> loadServerCredential
-  where
-    loadServerCredential :: IO T.Credential
-    loadServerCredential =
-      T.credentialLoadX509Chain certificateFile [caCertificateFile] privateKeyFile >>= \case
-        Right credential -> pure credential
-        Left _ -> putStrLn "invalid credential" >> exitFailure
-    fromCredential :: T.Credential -> T.ServerParams
-    fromCredential credential =
-      def
-        { T.serverWantClientCert = False,
-          T.serverShared = def {T.sharedCredentials = T.Credentials [credential]},
-          T.serverHooks = def,
-          T.serverSupported = supportedParameters
-        }
-
-loadFingerprint :: FilePath -> IO Fingerprint
-loadFingerprint certificateFile = do
-  (cert : _) <- SX.readSignedObject certificateFile
-  pure $ XV.getFingerprint (cert :: X.SignedExact X.Certificate) X.HashSHA256
-
--- * TLS 1.2 Transport
+-- * TLS Transport
 
 data TLS = TLS
   { tlsContext :: T.Context,
@@ -289,45 +178,21 @@ closeTLS ctx =
   (T.bye ctx >> T.contextClose ctx) -- sometimes socket was closed before 'TLS.bye'
     `E.catch` (\(_ :: E.SomeException) -> pure ()) -- so we catch the 'Broken pipe' error here
 
-mkTLSClientParams :: HostName -> ServiceName -> C.KeyHash -> T.ClientParams
-mkTLSClientParams host port keyHash = do
-  let p = B.pack port
-  (T.defaultParamsClient host p)
-    { T.clientShared = def,
-      T.clientHooks = def {T.onServerCertificate = \_ _ _ -> validateCertificateChain keyHash host p},
-      T.clientSupported = supportedParameters
-    }
-
-validateCertificateChain :: C.KeyHash -> HostName -> ByteString -> X.CertificateChain -> IO [XV.FailedReason]
-validateCertificateChain _ _ _ (X.CertificateChain []) = pure [XV.EmptyChain]
-validateCertificateChain _ _ _ (X.CertificateChain [_]) = pure [XV.EmptyChain]
-validateCertificateChain (C.KeyHash kh) host port cc@(X.CertificateChain sc@[_, caCert]) =
-  if Fingerprint kh == XV.getFingerprint caCert X.HashSHA256
-    then x509validate
-    else pure [XV.UnknownCA]
-  where
-    x509validate :: IO [XV.FailedReason]
-    x509validate = XV.validate X.HashSHA256 hooks checks certStore cache serviceID cc
-      where
-        hooks = XV.defaultHooks
-        checks = XV.defaultChecks
-        certStore = XS.makeCertificateStore sc
-        cache = XV.exceptionValidationCache [] -- we manually check fingerprint only of the identity certificate (ca.crt)
-        serviceID = (host, port)
-validateCertificateChain _ _ _ _ = pure [XV.AuthorityTooDeep]
-
 supportedParameters :: T.Supported
 supportedParameters =
   def
-    { T.supportedVersions = [T.TLS12],
-      T.supportedCiphers = [TE.cipher_ECDHE_ECDSA_CHACHA20POLY1305_SHA256],
+    { T.supportedVersions = [T.TLS13, T.TLS12],
+      T.supportedCiphers =
+        [ TE.cipher_TLS13_CHACHA20POLY1305_SHA256, -- for TLS13
+          TE.cipher_ECDHE_ECDSA_CHACHA20POLY1305_SHA256 -- for TLS12
+        ],
       T.supportedHashSignatures = [(T.HashIntrinsic, T.SignatureEd448), (T.HashIntrinsic, T.SignatureEd25519)],
       T.supportedSecureRenegotiation = False,
       T.supportedGroups = [T.X448, T.X25519]
     }
 
 instance Transport TLS where
-  transportName _ = "TLS 1.2"
+  transportName _ = "TLS"
   transportPeer = tlsPeer
   getServerConnection = getTLS TServer
   getClientConnection = getTLS TClient
@@ -424,9 +289,13 @@ data TransportError
   | -- | incorrect session ID
     TEBadSession
   | -- | transport handshake error
-    TEHandshake HandshakeError
+    TEHandshake {handshakeErr :: HandshakeError}
   deriving (Eq, Generic, Read, Show, Exception)
 
+instance ToJSON TransportError where
+  toJSON = J.genericToJSON . sumTypeJSON $ dropPrefix "TE"
+  toEncoding = J.genericToEncoding . sumTypeJSON $ dropPrefix "TE"
+
 -- | Transport handshake error.
 data HandshakeError
   = -- | parsing error
@@ -437,6 +306,10 @@ data HandshakeError
     IDENTITY
   deriving (Eq, Generic, Read, Show, Exception)
 
+instance ToJSON HandshakeError where
+  toJSON = J.genericToJSON $ sumTypeJSON id
+  toEncoding = J.genericToEncoding $ sumTypeJSON id
+
 instance Arbitrary TransportError where arbitrary = genericArbitraryU
 
 instance Arbitrary HandshakeError where arbitrary = genericArbitraryU

src/Simplex/Messaging/Transport/Client.hs

@@ -0,0 +1,82 @@
+{-# LANGUAGE ScopedTypeVariables #-}
+
+module Simplex.Messaging.Transport.Client
+  ( runTransportClient,
+    clientHandshake,
+  )
+where
+
+import Control.Monad.Except
+import Control.Monad.IO.Unlift
+import Data.ByteString.Char8 (ByteString)
+import qualified Data.ByteString.Char8 as B
+import Data.Default (def)
+import qualified Data.X509 as X
+import qualified Data.X509.CertificateStore as XS
+import Data.X509.Validation (Fingerprint (..))
+import qualified Data.X509.Validation as XV
+import GHC.IO.Exception (IOErrorType (..))
+import Network.Socket
+import qualified Network.TLS as T
+import qualified Simplex.Messaging.Crypto as C
+import Simplex.Messaging.Transport
+import System.IO.Error
+import UnliftIO.Exception (IOException)
+import qualified UnliftIO.Exception as E
+
+-- | Connect to passed TCP host:port and pass handle to the client.
+runTransportClient :: Transport c => MonadUnliftIO m => HostName -> ServiceName -> C.KeyHash -> (c -> m a) -> m a
+runTransportClient host port keyHash client = do
+  let clientParams = mkTLSClientParams host port keyHash
+  c <- liftIO $ startTCPClient host port clientParams
+  client c `E.finally` liftIO (closeConnection c)
+
+startTCPClient :: forall c. Transport c => HostName -> ServiceName -> T.ClientParams -> IO c
+startTCPClient host port clientParams = withSocketsDo $ resolve >>= tryOpen err
+  where
+    err :: IOException
+    err = mkIOError NoSuchThing "no address" Nothing Nothing
+
+    resolve :: IO [AddrInfo]
+    resolve =
+      let hints = defaultHints {addrSocketType = Stream}
+       in getAddrInfo (Just hints) (Just host) (Just port)
+
+    tryOpen :: IOException -> [AddrInfo] -> IO c
+    tryOpen e [] = E.throwIO e
+    tryOpen _ (addr : as) =
+      E.try (open addr) >>= either (`tryOpen` as) pure
+
+    open :: AddrInfo -> IO c
+    open addr = do
+      sock <- socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr)
+      connect sock $ addrAddress addr
+      ctx <- connectTLS clientParams sock
+      getClientConnection ctx
+
+mkTLSClientParams :: HostName -> ServiceName -> C.KeyHash -> T.ClientParams
+mkTLSClientParams host port keyHash = do
+  let p = B.pack port
+  (T.defaultParamsClient host p)
+    { T.clientShared = def,
+      T.clientHooks = def {T.onServerCertificate = \_ _ _ -> validateCertificateChain keyHash host p},
+      T.clientSupported = supportedParameters
+    }
+
+validateCertificateChain :: C.KeyHash -> HostName -> ByteString -> X.CertificateChain -> IO [XV.FailedReason]
+validateCertificateChain _ _ _ (X.CertificateChain []) = pure [XV.EmptyChain]
+validateCertificateChain _ _ _ (X.CertificateChain [_]) = pure [XV.EmptyChain]
+validateCertificateChain (C.KeyHash kh) host port cc@(X.CertificateChain sc@[_, caCert]) =
+  if Fingerprint kh == XV.getFingerprint caCert X.HashSHA256
+    then x509validate
+    else pure [XV.UnknownCA]
+  where
+    x509validate :: IO [XV.FailedReason]
+    x509validate = XV.validate X.HashSHA256 hooks checks certStore cache serviceID cc
+      where
+        hooks = XV.defaultHooks
+        checks = XV.defaultChecks
+        certStore = XS.makeCertificateStore sc
+        cache = XV.exceptionValidationCache [] -- we manually check fingerprint only of the identity certificate (ca.crt)
+        serviceID = (host, port)
+validateCertificateChain _ _ _ _ = pure [XV.AuthorityTooDeep]

src/Simplex/Messaging/Transport/Server.hs

@@ -0,0 +1,94 @@
+{-# LANGUAGE DuplicateRecordFields #-}
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+
+module Simplex.Messaging.Transport.Server
+  ( runTransportServer,
+    loadTLSServerParams,
+    loadFingerprint,
+    serverHandshake,
+  )
+where
+
+import Control.Monad.Except
+import Control.Monad.IO.Unlift
+import qualified Crypto.Store.X509 as SX
+import Data.Default (def)
+import Data.Set (Set)
+import qualified Data.Set as S
+import qualified Data.X509 as X
+import Data.X509.Validation (Fingerprint (..))
+import qualified Data.X509.Validation as XV
+import Network.Socket
+import qualified Network.TLS as T
+import Simplex.Messaging.Transport
+import System.Exit (exitFailure)
+import UnliftIO.Concurrent
+import qualified UnliftIO.Exception as E
+import UnliftIO.STM
+
+-- | Run transport server (plain TCP or WebSockets) on passed TCP port and signal when server started and stopped via passed TMVar.
+--
+-- All accepted connections are passed to the passed function.
+runTransportServer :: forall c m. (Transport c, MonadUnliftIO m) => TMVar Bool -> ServiceName -> T.ServerParams -> (c -> m ()) -> m ()
+runTransportServer started port serverParams server = do
+  u <- askUnliftIO
+  liftIO $ do
+    clients <- newTVarIO S.empty
+    E.bracket
+      (startTCPServer started port)
+      (closeServer clients)
+      $ \sock -> forever $ do
+        (connSock, _) <- accept sock
+        tid <- forkIO $ connectClient u connSock `E.catch` \(_ :: E.SomeException) -> pure ()
+        atomically . modifyTVar clients $ S.insert tid
+  where
+    connectClient :: UnliftIO m -> Socket -> IO ()
+    connectClient u connSock =
+      E.bracket
+        (connectTLS serverParams connSock >>= getServerConnection)
+        closeConnection
+        (unliftIO u . server)
+    closeServer :: TVar (Set ThreadId) -> Socket -> IO ()
+    closeServer clients sock = do
+      readTVarIO clients >>= mapM_ killThread
+      close sock
+      void . atomically $ tryPutTMVar started False
+
+startTCPServer :: TMVar Bool -> ServiceName -> IO Socket
+startTCPServer started port = withSocketsDo $ resolve >>= open >>= setStarted
+  where
+    resolve =
+      let hints = defaultHints {addrFlags = [AI_PASSIVE], addrSocketType = Stream}
+       in head <$> getAddrInfo (Just hints) Nothing (Just port)
+    open addr = do
+      sock <- socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr)
+      setSocketOption sock ReuseAddr 1
+      withFdSocket sock setCloseOnExecIfNeeded
+      bind sock $ addrAddress addr
+      listen sock 1024
+      return sock
+    setStarted sock = atomically (tryPutTMVar started True) >> pure sock
+
+loadTLSServerParams :: FilePath -> FilePath -> FilePath -> IO T.ServerParams
+loadTLSServerParams caCertificateFile certificateFile privateKeyFile =
+  fromCredential <$> loadServerCredential
+  where
+    loadServerCredential :: IO T.Credential
+    loadServerCredential =
+      T.credentialLoadX509Chain certificateFile [caCertificateFile] privateKeyFile >>= \case
+        Right credential -> pure credential
+        Left _ -> putStrLn "invalid credential" >> exitFailure
+    fromCredential :: T.Credential -> T.ServerParams
+    fromCredential credential =
+      def
+        { T.serverWantClientCert = False,
+          T.serverShared = def {T.sharedCredentials = T.Credentials [credential]},
+          T.serverHooks = def,
+          T.serverSupported = supportedParameters
+        }
+
+loadFingerprint :: FilePath -> IO Fingerprint
+loadFingerprint certificateFile = do
+  (cert : _) <- SX.readSignedObject certificateFile
+  pure $ XV.getFingerprint (cert :: X.SignedExact X.Certificate) X.HashSHA256

stack.yaml

@@ -37,16 +37,21 @@ packages:
 extra-deps:
   - cryptostore-0.2.1.0@sha256:9896e2984f36a1c8790f057fd5ce3da4cbcaf8aa73eb2d9277916886978c5b19,3881
   - simple-logger-0.1.0@sha256:be8ede4bd251a9cac776533bae7fb643369ebd826eb948a9a18df1a8dd252ff8,1079
+  - tls-1.5.7@sha256:1cc30253a9696b65a9cafc0317fbf09f7dcea15e3a145ed6c9c0e28c632fa23a,6991
+# below dependancies are to update Aeson to 2.0.3
+  - OneTuple-0.3.1@sha256:a848c096c9d29e82ffdd30a9998aa2931cbccb3a1bc137539d80f6174d31603e,2262
+  - attoparsec-0.14.4@sha256:79584bdada8b730cb5138fca8c35c76fbef75fc1d1e01e6b1d815a5ee9843191,5810
+  - hashable-1.4.0.2@sha256:0cddd0229d1aac305ea0404409c0bbfab81f075817bd74b8b2929eff58333e55,5005
+  - semialign-1.2.0.1@sha256:0e179b4d3a8eff79001d374d6c91917c6221696b9620f0a4d86852fc6a9b9501,2836
+  - text-short-0.1.5@sha256:962c6228555debdc46f758d0317dea16e5240d01419b42966674b08a5c3d8fa6,3498
+  - time-compat-1.9.6.1@sha256:42d8f2e08e965e1718917d54ad69e1d06bd4b87d66c41dc7410f59313dba4ed1,5033
+  - github: simplex-chat/aeson
+    commit: 3eb66f9a68f103b5f1489382aad89f5712a64db7
   # - ../hs-tls/core
-  - github: simplex-chat/hs-tls
-    commit: cea6d52c512716ff09adcac86ebc95bb0b3bb797
-    subdirs:
-      - core
-# - network-run-0.2.4@sha256:7dbb06def522dab413bce4a46af476820bffdff2071974736b06f52f4ab57c96,885
-# - git: https://github.com/commercialhaskell/stack.git
-#   commit: e7b331f14bcffb8367cd58fbfc8b40ec7642100a
-#
-# extra-deps: []
+  # - github: simplex-chat/hs-tls
+  #   commit: f6cc753611f80af300401cfae63846e9d7c40d9e
+  #   subdirs:
+  #     - core
 
 # Override default flag values for local packages and extra-deps
 # flags: {}

tests/AgentTests.hs

@@ -391,10 +391,10 @@ sendMessage (h1, name1) (h2, name2) msg = do
   ("m1", name2', Right (MID mId)) <- h1 #: ("m1", name2, "SEND :" <> msg)
   name2' `shouldBe` name2
   h1 <#= \case ("", n, SENT m) -> n == name2 && m == mId; _ -> False
-  ("", name1', Right (MSG MsgMeta {recipient = (msgId, _)} msg')) <- (h2 <#:)
+  ("", name1', Right (MSG MsgMeta {recipient = (msgId', _)} msg')) <- (h2 <#:)
   name1' `shouldBe` name1
   msg' `shouldBe` msg
-  h2 #: ("m2", name1, "ACK " <> bshow msgId) =#> \case ("m2", n, OK) -> n == name1; _ -> False
+  h2 #: ("m2", name1, "ACK " <> bshow msgId') =#> \case ("m2", n, OK) -> n == name1; _ -> False
 
 -- connect' :: forall c. Transport c => c -> c -> IO (ByteString, ByteString)
 -- connect' h1 h2 = do

tests/AgentTests/FunctionalAPITests.hs

@@ -12,7 +12,7 @@ import Control.Monad.IO.Unlift
 import SMPAgentClient
 import SMPClient (withSmpServer)
 import Simplex.Messaging.Agent
-import Simplex.Messaging.Agent.Env.SQLite (dbFile)
+import Simplex.Messaging.Agent.Env.SQLite (AgentConfig (..))
 import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Protocol (ErrorType (..), MsgBody)
 import Simplex.Messaging.Transport (ATransport (..))
@@ -44,6 +44,8 @@ functionalAPITests t = do
       withSmpServer t testAsyncJoiningOfflineBeforeActivation
     it "should connect with both clients going offline" $
       withSmpServer t testAsyncBothOffline
+    it "should notify after HELLO timeout" $
+      withSmpServer t testAsyncHelloTimeout
 
 testAgentClient :: IO ()
 testAgentClient = do
@@ -146,6 +148,17 @@ testAsyncBothOffline = do
     exchangeGreetings alice' bobId bob' aliceId
   pure ()
 
+testAsyncHelloTimeout :: IO ()
+testAsyncHelloTimeout = do
+  alice <- getSMPAgentClient cfg
+  bob <- getSMPAgentClient cfg {dbFile = testDB2, helloTimeout = 1}
+  Right () <- runExceptT $ do
+    (_, cReq) <- createConnection alice SCMInvitation
+    disconnectAgentClient alice
+    aliceId <- joinConnection bob cReq "bob's connInfo"
+    get bob ##> ("", aliceId, ERR $ CONN NOT_ACCEPTED)
+  pure ()
+
 exchangeGreetings :: AgentClient -> ConnId -> AgentClient -> ConnId -> ExceptT AgentErrorType IO ()
 exchangeGreetings alice bobId bob aliceId = do
   4 <- sendMessage alice bobId "hello"

tests/CoreTests/ProtocolErrorTests.hs

@@ -1,8 +1,9 @@
 module CoreTests.ProtocolErrorTests where
 
-import Simplex.Messaging.Agent.Protocol (AgentErrorType, agentErrorTypeP, serializeAgentError, serializeSmpErrorType, smpErrorTypeP)
+import Simplex.Messaging.Agent.Protocol (AgentErrorType)
 import Simplex.Messaging.Parsers (parseAll)
 import Simplex.Messaging.Protocol (ErrorType)
+import Simplex.Messaging.Encoding.String
 import Test.Hspec
 import Test.Hspec.QuickCheck (modifyMaxSuccess)
 import Test.QuickCheck
@@ -11,8 +12,8 @@ protocolErrorTests :: Spec
 protocolErrorTests = modifyMaxSuccess (const 1000) $ do
   describe "errors parsing / serializing" $ do
     it "should parse SMP protocol errors" . property $ \err ->
-      parseAll smpErrorTypeP (serializeSmpErrorType err)
+      parseAll strP (strEncode err)
         == Right (err :: ErrorType)
     it "should parse SMP agent errors" . property $ \err ->
-      parseAll agentErrorTypeP (serializeAgentError err)
+      parseAll strP (strEncode err)
         == Right (err :: AgentErrorType)

tests/SMPAgentClient.hs

@@ -20,12 +20,13 @@ import SMPClient
     withSmpServerOn,
     withSmpServerThreadOn,
   )
-import Simplex.Messaging.Agent (runSMPAgentBlocking)
 import Simplex.Messaging.Agent.Env.SQLite
 import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Agent.RetryInterval
+import Simplex.Messaging.Agent.Server (runSMPAgentBlocking)
 import Simplex.Messaging.Client (SMPClientConfig (..), smpDefaultConfig)
 import Simplex.Messaging.Transport
+import Simplex.Messaging.Transport.Client
 import Test.Hspec
 import UnliftIO.Concurrent
 import UnliftIO.Directory

tests/SMPClient.hs

@@ -21,6 +21,7 @@ import Simplex.Messaging.Server (runSMPServerBlocking)
 import Simplex.Messaging.Server.Env.STM
 import Simplex.Messaging.Server.StoreLog (openReadStoreLog)
 import Simplex.Messaging.Transport
+import Simplex.Messaging.Transport.Client
 import Test.Hspec
 import UnliftIO.Concurrent
 import qualified UnliftIO.Exception as E

tests/Test.hs

@@ -18,7 +18,7 @@ main = do
       describe "Encoding tests" encodingTests
       describe "Protocol error tests" protocolErrorTests
       describe "Version range" versionRangeTests
-    describe "SMP server via TLS 1.3" $ serverTests (transport @TLS)
+    describe "SMP server via TLS" $ serverTests (transport @TLS)
     describe "SMP server via WebSockets" $ serverTests (transport @WS)
     describe "SMP client agent" $ agentTests (transport @TLS)
   removeDirectoryRecursive "tests/tmp"