Merge branch 'queue-journals' into ep/qj-rename

.github/workflows/docker-image.yml

@@ -14,22 +14,22 @@ jobs:
       matrix:
         include:
           - app: smp-server
-            app_port: 5223
+            app_port: "443 5223"
           - app: xftp-server
-            app_port: 443 
+            app_port: 443
     steps:
       - name: Clone project
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
 
       - name: Log in to Docker Hub
-        uses: docker/login-action@v2
+        uses: docker/login-action@v3
         with:
           username: ${{ secrets.DOCKERHUB_USERNAME }}
           password: ${{ secrets.DOCKERHUB_PASSWORD }}
 
       - name: Extract metadata for Docker image
         id: meta
-        uses: docker/metadata-action@v4
+        uses: docker/metadata-action@v5
         with:
           images: ${{ secrets.DOCKERHUB_USERNAME }}/${{ matrix.app }}
           flavor: |
@@ -40,7 +40,7 @@ jobs:
             type=semver,pattern=v{{major}}
 
       - name: Build and push Docker image
-        uses: docker/build-push-action@v4
+        uses: docker/build-push-action@v6
         with:
           push: true
           build-args: |

Dockerfile

@@ -1,15 +1,20 @@
-ARG TAG=22.04
+# syntax=docker/dockerfile:1.7.0-labs
+ARG TAG=24.04
 
 FROM ubuntu:${TAG} AS build
 
 ### Build stage
 
 # Install curl and git and simplexmq dependencies
-RUN apt-get update && apt-get install -y curl git build-essential libgmp3-dev zlib1g-dev llvm-12 llvm-12-dev libnuma-dev libssl-dev
+RUN apt-get update && apt-get install -y curl git build-essential libgmp3-dev zlib1g-dev llvm-18 llvm-18-dev libnuma-dev libssl-dev
 
 # Specify bootstrap Haskell versions
 ENV BOOTSTRAP_HASKELL_GHC_VERSION=9.6.3
-ENV BOOTSTRAP_HASKELL_CABAL_VERSION=3.10.1.0
+ENV BOOTSTRAP_HASKELL_CABAL_VERSION=3.12.1.0
+
+# Do not install Stack
+ENV BOOTSTRAP_HASKELL_INSTALL_NO_STACK=true
+ENV BOOTSTRAP_HASKELL_INSTALL_NO_STACK_HOOK=true
 
 # Install ghcup
 RUN curl --proto '=https' --tlsv1.2 -sSf https://get-ghcup.haskell.org | BOOTSTRAP_HASKELL_NONINTERACTIVE=1 sh
@@ -21,26 +26,42 @@ ENV PATH="/root/.cabal/bin:/root/.ghcup/bin:$PATH"
 RUN ghcup set ghc "${BOOTSTRAP_HASKELL_GHC_VERSION}" && \
     ghcup set cabal "${BOOTSTRAP_HASKELL_CABAL_VERSION}"
 
-COPY . /project
+# Copy only the source code
+COPY apps /project/apps/
+COPY cbits /project/cbits/
+COPY src /project/src/
+
+COPY cabal.project Setup.hs simplexmq.cabal LICENSE /project
+
 WORKDIR /project
 
+# Debug
+#ARG CACHEBUST=1
+
+#ADD --chmod=755 https://github.com/MShekow/directory-checksum/releases/download/v1.4.6/directory-checksum_1.4.6_linux_amd64 /usr/local/bin/directory-checksum
+#RUN directory-checksum --max-depth 2 .
+
+# Set build arguments and check if they exist
 ARG APP
-ARG APP_PORT
-RUN if [ -z "$APP" ] || [ -z "$APP_PORT" ]; then printf "Please spcify \$APP and \$APP_PORT build-arg.\n"; exit 1; fi
+RUN if [ -z "$APP" ]; then printf "Please spcify \$APP build-arg.\n"; exit 1; fi
 
 # Compile app
 RUN cabal update
 RUN cabal build exe:$APP
 
+# Copy scripts
+COPY scripts /project/scripts/
+
 # Create new path containing all files needed
 RUN mkdir /final
 WORKDIR /final
 
 # Strip the binary from debug symbols to reduce size
-RUN bin=$(find /project/dist-newstyle -name "$APP" -type f -executable) && \
+RUN bin="$(find /project/dist-newstyle -name "$APP" -type f -executable)" && \
     mv "$bin" ./ && \
     strip ./"$APP" &&\
-    mv /project/scripts/docker/entrypoint-"$APP" ./entrypoint
+    mv /project/scripts/docker/entrypoint-"$APP" ./entrypoint &&\
+    mv /project/scripts/main/simplex-servers-stopscript ./simplex-servers-stopscript
 
 ### Final stage
 FROM ubuntu:${TAG}
@@ -53,6 +74,8 @@ COPY --from=build /final /usr/local/bin/
 
 # Open app listening port
 ARG APP_PORT
+RUN if [ -z "$APP_PORT" ]; then printf "Please spcify \$APP_PORT build-arg.\n"; exit 1; fi
+
 EXPOSE $APP_PORT
 
 # simplexmq requires using SIGINT to correctly preserve undelivered messages and restore them on restart

scripts/docker/docker-compose-smp-complete.yml

@@ -0,0 +1,67 @@
+name: SimpleX Chat - smp-server
+
+services:
+  oneshot:
+    image: ubuntu:latest
+    environment:
+      CADDYCONF: |
+        ${CADDY_OPTS:-}
+
+        http://{$$ADDR} {
+            redir https://{$$ADDR}{uri} permanent
+        }
+
+        {$$ADDR}:8443 {
+            tls {
+                key_type rsa4096
+            }
+        }
+    command: sh -c 'if [ ! -f /etc/caddy/Caddyfile ]; then printf "$${CADDYCONF}" > /etc/caddy/Caddyfile; fi'
+    volumes:
+      - ./caddy_conf:/etc/caddy
+
+  caddy:
+    image: caddy:latest
+    depends_on:
+      oneshot:
+        condition: service_completed_successfully
+    cap_add:
+      - NET_ADMIN
+    environment:
+      ADDR: ${ADDR?"Please specify the domain."}
+    volumes:
+      - ./caddy_conf:/etc/caddy
+      - caddy_data:/data
+      - caddy_config:/config
+    ports:
+      - 80:80
+    restart: unless-stopped
+    healthcheck:
+      test: "test -d /data/caddy/certificates/${CERT_PATH:-acme-v02.api.letsencrypt.org-directory}/${ADDR} || exit 1"
+      interval: 1s
+      retries: 60
+
+  smp-server:
+    image: ${SIMPLEX_IMAGE:-simplexchat/smp-server:latest}
+    depends_on:
+      caddy:
+        condition: service_healthy
+    environment:
+      ADDR: ${ADDR?"Please specify the domain."}
+      PASS: ${PASS:-}
+    volumes:
+      - ./smp_configs:/etc/opt/simplex
+      - ./smp_state:/var/opt/simplex
+      - type: volume
+        source: caddy_data
+        target: /certificates
+        volume:
+          subpath: "caddy/certificates/${CERT_PATH:-acme-v02.api.letsencrypt.org-directory}/${ADDR}"
+    ports:
+      - 443:443
+      - 5223:5223
+    restart: unless-stopped
+
+volumes:
+  caddy_data:
+  caddy_config:

scripts/docker/docker-compose-smp-manual.yml

@@ -0,0 +1,15 @@
+name: SimpleX Chat - smp-server
+
+services:
+  smp-server:
+    image: ${SIMPLEX_IMAGE:-simplexchat/smp-server:latest}
+    environment:
+      WEB_MANUAL: ${WEB_MANUAL:-1}
+      ADDR: ${ADDR?"Please specify the domain."}
+      PASS: ${PASS:-}
+    volumes:
+      - ./smp_configs:/etc/opt/simplex
+      - ./smp_state:/var/opt/simplex
+    ports:
+      - 5223:5223
+    restart: unless-stopped

scripts/docker/docker-compose-smp.env

@@ -0,0 +1,11 @@
+# Mandatory
+ADDR=your_ip_or_addr
+
+# Optional
+#PASS='123123'
+#WEB_MANUAL=1
+
+# Debug
+#SIMPLEX_SMP_IMAGE=smp-server-dev
+#CERT_PATH=acme-staging-v02.api.letsencrypt.org-directory
+#CADDY_OPTS='{\n    acme_ca https://acme-staging-v02.api.letsencrypt.org/directory\n}'

scripts/docker/docker-compose-xftp.env

@@ -0,0 +1,9 @@
+# Mandatory
+ADDR=your_ip_or_addr
+QUOTA=120gb
+
+# Optional
+#PASS='123123'
+
+# Debug
+#SIMPLEX_XFTP_IMAGE=xftp-server-dev

scripts/docker/docker-compose-xftp.yml

@@ -0,0 +1,16 @@
+name: SimpleX Chat - xftp-server
+
+services:
+  xftp-server:
+    image: ${SIMPLEX_XFTP_IMAGE:-simplexchat/xftp-server:latest}
+    environment:
+      ADDR: ${ADDR?"Please specify the domain."}
+      QUOTA: ${QUOTA?"Please specify disk quota."}
+      PASS: ${PASS:-}
+    volumes:
+      - ./xftp_configs:/etc/opt/simplex-xftp
+      - ./xftp_state:/var/opt/simplex-xftp
+      - ./xftp_files:/srv/xftp
+    ports:
+      - 443:443
+    restart: unless-stopped

scripts/docker/entrypoint-smp-server

@@ -1,48 +1,87 @@
 #!/usr/bin/env sh
+set -e
+
 confd='/etc/opt/simplex'
-logd='/var/opt/simplex/'
+cert_path='/certificates'
 
 # Check if server has been initialized
 if [ ! -f "${confd}/smp-server.ini" ]; then
   # If not, determine ip or domain
   case "${ADDR}" in
-    '') printf 'Please specify $ADDR environment variable.\n'; exit 1 ;;
+    '')
+      printf 'Please specify $ADDR environment variable.\n'
+      exit 1
+      ;;
+
+    # Determine domain or IPv6
     *[a-zA-Z]*)
       case "${ADDR}" in
-        *:*) set -- --ip "${ADDR}" ;;
-        *) set -- -n "${ADDR}" ;;
+        # IPv6
+        *:*)
+          set -- --ip "${ADDR}"
+          ;;
+
+        # Domain
+        *)
+          case "${ADDR}" in
+            # It's in domain format
+            *.*)
+              # Determine the base domain
+              ADDR_BASE="$(printf '%s' "$ADDR" | awk -F. '{print $(NF-1)"."$NF}')"
+              set -- --fqdn "${ADDR}" --own-domains="${ADDR_BASE}"
+              ;;
+
+            # Incorrect domain
+            *)
+              printf 'Incorrect $ADDR environment variable. Please specify the correct one in format: smp1.example.org / example.org \n'
+              exit 1
+              ;;
+          esac
       esac
       ;;
-    *) set -- --ip "${ADDR}" ;;
+
+    # Assume everything else is IPv4
+    *)
+      set -- --ip "${ADDR}" ;;
   esac
 
   # Optionally, set password
   case "${PASS}" in
-    '') set -- "$@" --no-password ;;
-    *) set -- "$@" --password "${PASS}" ;;
+    # Empty value = no password
+    '')
+      set -- "$@" --no-password
+      ;;
+
+    # Assume that everything else is a password
+    *)
+      set -- "$@" --password "${PASS}"
+      ;;
   esac
 
   # And init certificates and configs
-  smp-server init -y -l "$@"
+  smp-server init --yes \
+                  --store-log \
+                  --daily-stats \
+                  --source-code \
+                  "$@" > /dev/null 2>&1
+
+  # Fix path to certificates
+  if [ -n "${WEB_MANUAL}" ]; then
+    sed -i -e 's|^[^#]*https: |#&|' \
+           -e 's|^[^#]*cert: |#&|' \
+           -e 's|^[^#]*key: |#&|' \
+           -e 's|^port:.*|port: 5223|' \
+           "${confd}/smp-server.ini"
+  else
+    sed -i -e "s|cert: /etc/opt/simplex/web.crt|cert: $cert_path/$ADDR.crt|" \
+           -e "s|key: /etc/opt/simplex/web.key|key: $cert_path/$ADDR.key|" \
+           "${confd}/smp-server.ini"
+  fi
 fi
 
 # Backup store log just in case
-#
-# Uses the UTC (universal) time zone and this
-# format: YYYY-mm-dd'T'HH:MM:SS
-# year, month, day, letter T, hour, minute, second
-#
-# This is the ISO 8601 format without the time zone at the end.
-#
-_file="${logd}/smp-server-store.log"
-if [ -f "${_file}" ]; then
-  _backup_extension="$(date -u '+%Y-%m-%dT%H:%M:%S')"
-  cp -v -p "${_file}" "${_file}.${_backup_extension:-date-failed}"
-  unset -v _backup_extension
-fi
-unset -v _file
+DOCKER=true /usr/local/bin/simplex-servers-stopscript smp-server
 
 # Finally, run smp-sever. Notice that "exec" here is important:
 # smp-server replaces our helper script, so that it can catch INT signal
 exec smp-server start +RTS -N -RTS
-

scripts/docker/entrypoint-xftp-server

@@ -1,50 +1,90 @@
 #!/usr/bin/env sh
+set -eu
+
 confd='/etc/opt/simplex-xftp'
-logd='/var/opt/simplex-xftp'
 
 # Check if server has been initialized
 if [ ! -f "${confd}/file-server.ini" ]; then
   # If not, determine ip or domain
   case "${ADDR}" in
-    '') printf 'Please specify $ADDR environment variable.\n'; exit 1 ;;
+    '')
+      printf 'Please specify $ADDR environment variable.\n'
+      exit 1
+      ;;
+
+    # Determine domain or IPv6
     *[a-zA-Z]*)
       case "${ADDR}" in
-        *:*) set -- --ip "${ADDR}" ;;
-        *) set -- -n "${ADDR}" ;;
+        # IPv6
+        *:*)
+          set -- --ip "${ADDR}"
+          ;;
+
+        # Domain
+        *)
+          case "${ADDR}" in
+            # Check if format is correct
+            *.*)
+              set -- --fqdn "${ADDR}"
+              ;;
+
+            # Incorrect domain
+            *)
+              printf 'Incorrect $ADDR environment variable. Please specify the correct one in format: smp1.example.org / example.org \n'
+              exit 1
+              ;;
+          esac
+          ;;
       esac
       ;;
-    *) set -- --ip "${ADDR}" ;;
+
+    # Assume everything else is IPv4
+    *)
+      set -- --ip "${ADDR}"
+      ;;
   esac
 
-  # Set quota
+  # Set global disk quota
   case "${QUOTA}" in
-    '') printf 'Please specify $QUOTA environment variable.\n'; exit 1 ;;
-    *GB) QUOTA="$(printf ${QUOTA} | tr '[:upper:]' '[:lower:]')"; set -- "$@" --quota "${QUOTA}" ;;
-    *gb) set -- "$@" --quota "${QUOTA}" ;;
-    *) printf 'Wrong format. Format should be: 1gb, 10gb, 100gb.\n'; exit 1 ;;
+    '')
+      printf 'Please specify $QUOTA environment variable.\n'
+      exit 1
+      ;;
+
+    # Incorrect format in uppercase, but automagically workaround this, replacing characters to lowercase
+    *GB)
+      QUOTA="$(printf '%s' "${QUOTA}" | tr '[:upper:]' '[:lower:]')"
+      set -- "$@" --quota "${QUOTA}"
+      ;;
+
+    # Correct format  
+    *gb)
+      set -- "$@" --quota "${QUOTA}"
+      ;;
+
+    # Incorrect format
+    *)
+      printf 'Wrong format. Format should be: 1gb, 10gb, 100gb.\n'
+      exit 1
+      ;;
   esac
 
   # Init the certificates and configs
-  xftp-server init -l -p /srv/xftp "$@"
+  xftp-server init --store-log \
+                   --path /srv/xftp \
+                   "$@" > /dev/null 2>&1
+
+  # Optionally, set password
+  if [ -n "${PASS}" ]; then
+    sed -i -e "/^# create_password:/a create_password: $PASS" \
+           "${confd}/file-server.ini"
+  fi
 fi
 
 # Backup store log just in case
-#
-# Uses the UTC (universal) time zone and this
-# format: YYYY-mm-dd'T'HH:MM:SS
-# year, month, day, letter T, hour, minute, second
-#
-# This is the ISO 8601 format without the time zone at the end.
-#
-_file="${logd}/file-server-store.log"
-if [ -f "${_file}" ]; then
-  _backup_extension="$(date -u '+%Y-%m-%dT%H:%M:%S')"
-  cp -v -p "${_file}" "${_file}.${_backup_extension:-date-failed}"
-  unset -v _backup_extension
-fi
-unset -v _file
+
+DOCKER=true /usr/local/bin/simplex-servers-stopscript xftp-server
 
 # Finally, run xftp-sever. Notice that "exec" here is important:
 # smp-server replaces our helper script, so that it can catch INT signal
 exec xftp-server start +RTS -N -RTS
-

scripts/main/simplex-servers-stopscript

@@ -148,8 +148,10 @@ xftp_cleanup() {
 
 main() {
 	type="${1:-}"
-  
-	checks
+
+  if [ -z "${DOCKER+x}" ]; then
+  	checks
+  fi
 
 	case "$type" in
 		smp-server)

scripts/main/smp-server.service

@@ -5,12 +5,21 @@ Description=SMP server
 User=smp
 Group=smp
 Type=simple
+
 ExecStart=/usr/local/bin/smp-server start +RTS -N -RTS 
 ExecStopPost=/usr/local/bin/simplex-servers-stopscript smp-server
+
 LimitNOFILE=65535
 KillSignal=SIGINT
+
 TimeoutStartSec=infinity
 TimeoutStopSec=infinity
+
+Restart=on-failure
+RestartSec=10s
+StartLimitBurst=3
+StartLimitInterval=60s
+
 AmbientCapabilities=CAP_NET_BIND_SERVICE
 
 [Install]

scripts/main/xftp-server.service

@@ -5,12 +5,21 @@ Description=XFTP server
 User=xftp
 Group=xftp
 Type=simple
+
 ExecStart=/usr/local/bin/xftp-server start +RTS -N -RTS
 ExecStopPost=/usr/local/bin/simplex-servers-stopscript xftp-server
+
 LimitNOFILE=65535
 KillSignal=SIGINT
+
 TimeoutStartSec=infinity
 TimeoutStopSec=infinity
+
+Restart=on-failure
+RestartSec=10s
+StartLimitBurst=3
+StartLimitInterval=60s
+
 AmbientCapabilities=CAP_NET_BIND_SERVICE
    
 [Install]

simplexmq.cabal

@@ -1,7 +1,7 @@
 cabal-version: 1.12
 
 name:           simplexmq
-version:        6.3.0.2
+version:        6.3.0.3
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and

src/Simplex/FileTransfer/Description.hs

@@ -1,4 +1,3 @@
-{-# LANGUAGE CPP #-}
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DeriveAnyClass #-}
 {-# LANGUAGE DerivingStrategies #-}
@@ -71,20 +70,13 @@ import qualified Data.Yaml as Y
 import Simplex.FileTransfer.Chunks
 import Simplex.FileTransfer.Protocol
 import Simplex.Messaging.Agent.QueryString
-import Simplex.Messaging.Agent.Store.DB (Binary (..))
+import Simplex.Messaging.Agent.Store.DB (Binary (..), FromField (..), ToField (..))
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (defaultJSON, parseAll)
 import Simplex.Messaging.Protocol (XFTPServer)
 import Simplex.Messaging.ServiceScheme (ServiceScheme (..))
 import Simplex.Messaging.Util (bshow, safeDecodeUtf8, (<$?>))
-#if defined(dbPostgres)
-import Database.PostgreSQL.Simple.FromField (FromField (..))
-import Database.PostgreSQL.Simple.ToField (ToField (..))
-#else
-import Database.SQLite.Simple.FromField (FromField (..))
-import Database.SQLite.Simple.ToField (ToField (..))
-#endif
 
 data FileDescription (p :: FileParty) = FileDescription
   { party :: SFileParty p,

src/Simplex/FileTransfer/Types.hs

@@ -1,4 +1,3 @@
-{-# LANGUAGE CPP #-}
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
@@ -23,13 +22,7 @@ import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers
 import Simplex.Messaging.Protocol (XFTPServer)
 import System.FilePath ((</>))
-#if defined(dbPostgres)
-import Database.PostgreSQL.Simple.FromField (FromField (..))
-import Database.PostgreSQL.Simple.ToField (ToField (..))
-#else
-import Database.SQLite.Simple.FromField (FromField (..))
-import Database.SQLite.Simple.ToField (ToField (..))
-#endif
+import Simplex.Messaging.Agent.Store.DB (FromField (..), ToField (..))
 
 type RcvFileId = ByteString -- Agent entity ID
 

src/Simplex/Messaging/Agent.hs

@@ -1805,8 +1805,8 @@ prepareDeleteConnections_ getConnections c waitDelivery connIds = do
   -- ! if it was used to notify about the result, it might be necessary to differentiate
   -- ! between completed deletions of connections, and deletions delayed due to wait for delivery (see deleteConn)
   deliveryTimeout <- if waitDelivery then asks (Just . connDeleteDeliveryTimeout . config) else pure Nothing
-  rs' <- lift $ catMaybes . rights <$> withStoreBatch' c (\db -> map (deleteConn db deliveryTimeout) (M.keys delRs))
-  forM_ rs' $ \cId -> notify ("", cId, AEvt SAEConn DEL_CONN)
+  cIds_ <- lift $ L.nonEmpty . catMaybes . rights <$> withStoreBatch' c (\db -> map (deleteConn db deliveryTimeout) (M.keys delRs))
+  forM_ cIds_ $ \cIds -> notify ("", "", AEvt SAEConn $ DEL_CONNS cIds)
   pure (errs' <> delRs, rqs, connIds')
   where
     rcvQueues :: SomeConn -> Either (Either AgentErrorType ()) [RcvQueue]
@@ -1826,32 +1826,33 @@ deleteConnQueues c waitDelivery ntf rqs = do
   rs <- connResults <$> (deleteQueueRecs =<< deleteQueues c rqs)
   let connIds = M.keys $ M.filter isRight rs
   deliveryTimeout <- if waitDelivery then asks (Just . connDeleteDeliveryTimeout . config) else pure Nothing
-  rs' <- catMaybes . rights <$> withStoreBatch' c (\db -> map (deleteConn db deliveryTimeout) connIds)
-  forM_ rs' $ \cId -> notify ("", cId, AEvt SAEConn DEL_CONN)
+  cIds_ <- L.nonEmpty . catMaybes . rights <$> withStoreBatch' c (\db -> map (deleteConn db deliveryTimeout) connIds)
+  forM_ cIds_ $ \cIds -> notify ("", "", AEvt SAEConn $ DEL_CONNS cIds)
   pure rs
   where
     deleteQueueRecs :: [(RcvQueue, Either AgentErrorType ())] -> AM' [(RcvQueue, Either AgentErrorType ())]
     deleteQueueRecs rs = do
       maxErrs <- asks $ deleteErrorCount . config
-      (rs', notifyActions) <- unzip . rights <$> withStoreBatch' c (\db -> map (deleteQueueRec db maxErrs) rs)
-      mapM_ sequence_ notifyActions
-      pure rs'
+      rs' <- rights <$> withStoreBatch' c (\db -> map (deleteQueueRec db maxErrs) rs)
+      let delQ ((rq, _), err_) =  (qConnId rq,qServer rq,queueId rq,) <$> err_
+          delQs_ = L.nonEmpty $ mapMaybe delQ rs'
+      forM_ delQs_ $ \delQs -> notify ("", "", AEvt SAEConn $ DEL_RCVQS delQs)
+      pure $ map fst rs'
       where
         deleteQueueRec ::
           DB.Connection ->
           Int ->
           (RcvQueue, Either AgentErrorType ()) ->
-          IO ((RcvQueue, Either AgentErrorType ()), Maybe (AM' ()))
+          IO ((RcvQueue, Either AgentErrorType ()), Maybe (Maybe AgentErrorType)) -- Nothing - no event, Just Nothing - no error
         deleteQueueRec db maxErrs (rq@RcvQueue {userId, server}, r) = case r of
-          Right _ -> deleteConnRcvQueue db rq $> ((rq, r), Just (notifyRQ rq Nothing))
+          Right _ -> deleteConnRcvQueue db rq $> ((rq, r), Just Nothing)
           Left e
             | temporaryOrHostError e && deleteErrors rq + 1 < maxErrs -> incRcvDeleteErrors db rq $> ((rq, r), Nothing)
             | otherwise -> do
                 deleteConnRcvQueue db rq
                 -- attempts and successes are counted in deleteQueues function
                 atomically $ incSMPServerStat c userId server connDeleted
-                pure ((rq, Right ()), Just (notifyRQ rq (Just e)))
-    notifyRQ rq e_ = notify ("", qConnId rq, AEvt SAEConn $ DEL_RCVQ (qServer rq) (queueId rq) e_)
+                pure ((rq, Right ()), Just (Just e))
     notify = when ntf . atomically . writeTBQueue (subQ c)
     connResults :: [(RcvQueue, Either AgentErrorType ())] -> Map ConnId (Either AgentErrorType ())
     connResults = M.map snd . foldl' addResult M.empty

src/Simplex/Messaging/Agent/Protocol.hs

@@ -1,4 +1,3 @@
-{-# LANGUAGE CPP #-}
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DeriveAnyClass #-}
 {-# LANGUAGE DuplicateRecordFields #-}
@@ -168,12 +167,12 @@ import Data.Time.Clock.System (SystemTime)
 import Data.Type.Equality
 import Data.Typeable ()
 import Data.Word (Word16, Word32)
+import Simplex.Messaging.Agent.Store.DB (FromField (..), ToField (..))
 import Simplex.FileTransfer.Description
 import Simplex.FileTransfer.Protocol (FileParty (..))
 import Simplex.FileTransfer.Transport (XFTPErrorType)
 import Simplex.FileTransfer.Types (FileErrorType)
 import Simplex.Messaging.Agent.QueryString
-import Simplex.Messaging.Agent.Store.DB (Binary (..))
 import Simplex.Messaging.Client (ProxyClientError)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Crypto.Ratchet
@@ -224,13 +223,6 @@ import Simplex.Messaging.Version
 import Simplex.Messaging.Version.Internal
 import Simplex.RemoteControl.Types
 import UnliftIO.Exception (Exception)
-#if defined(dbPostgres)
-import Database.PostgreSQL.Simple.FromField (FromField (..))
-import Database.PostgreSQL.Simple.ToField (ToField (..))
-#else
-import Database.SQLite.Simple.FromField (FromField (..))
-import Database.SQLite.Simple.ToField (ToField (..))
-#endif
 
 -- SMP agent protocol version history:
 -- 1 - binary protocol encoding (1/1/2022)
@@ -366,8 +358,8 @@ data AEvent (e :: AEntity) where
   MSGNTF :: MsgId -> Maybe UTCTime -> AEvent AEConn
   RCVD :: MsgMeta -> NonEmpty MsgReceipt -> AEvent AEConn
   QCONT :: AEvent AEConn
-  DEL_RCVQ :: SMPServer -> SMP.RecipientId -> Maybe AgentErrorType -> AEvent AEConn
-  DEL_CONN :: AEvent AEConn
+  DEL_RCVQS :: NonEmpty (ConnId, SMPServer, SMP.RecipientId, Maybe AgentErrorType) -> AEvent AEConn
+  DEL_CONNS :: NonEmpty ConnId -> AEvent AEConn
   DEL_USER :: Int64 -> AEvent AENone
   STAT :: ConnectionStats -> AEvent AEConn
   OK :: AEvent AEConn
@@ -437,8 +429,8 @@ data AEventTag (e :: AEntity) where
   MSGNTF_ :: AEventTag AEConn
   RCVD_ :: AEventTag AEConn
   QCONT_ :: AEventTag AEConn
-  DEL_RCVQ_ :: AEventTag AEConn
-  DEL_CONN_ :: AEventTag AEConn
+  DEL_RCVQS_ :: AEventTag AEConn
+  DEL_CONNS_ :: AEventTag AEConn
   DEL_USER_ :: AEventTag AENone
   STAT_ :: AEventTag AEConn
   OK_ :: AEventTag AEConn
@@ -492,8 +484,8 @@ aEventTag = \case
   MSGNTF {} -> MSGNTF_
   RCVD {} -> RCVD_
   QCONT -> QCONT_
-  DEL_RCVQ {} -> DEL_RCVQ_
-  DEL_CONN -> DEL_CONN_
+  DEL_RCVQS _ -> DEL_RCVQS_
+  DEL_CONNS _ -> DEL_CONNS_
   DEL_USER _ -> DEL_USER_
   STAT _ -> STAT_
   OK -> OK_
@@ -651,7 +643,7 @@ instance ToJSON NotificationsMode where
 instance FromJSON NotificationsMode where
   parseJSON = strParseJSON "NotificationsMode"
 
-instance ToField NotificationsMode where toField = toField . Binary . strEncode
+instance ToField NotificationsMode where toField = toField . strEncode
 
 instance FromField NotificationsMode where fromField = blobFieldDecoder $ parseAll strP
 

src/Simplex/Messaging/Agent/Stats.hs

@@ -1,4 +1,3 @@
-{-# LANGUAGE CPP #-}
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE GeneralizedNewtypeDeriving #-}
 {-# LANGUAGE NamedFieldPuns #-}
@@ -12,17 +11,11 @@ import Data.Int (Int64)
 import Data.Map.Strict (Map)
 import qualified Data.Map.Strict as M
 import Simplex.Messaging.Agent.Protocol (UserId)
+import Simplex.Messaging.Agent.Store.DB (FromField (..), ToField (..))
 import Simplex.Messaging.Parsers (defaultJSON, fromTextField_)
 import Simplex.Messaging.Protocol (NtfServer, SMPServer, XFTPServer)
 import Simplex.Messaging.Util (decodeJSON, encodeJSON)
 import UnliftIO.STM
-#if defined(dbPostgres)
-import Database.PostgreSQL.Simple.FromField (FromField (..))
-import Database.PostgreSQL.Simple.ToField (ToField (..))
-#else
-import Database.SQLite.Simple.FromField (FromField (..))
-import Database.SQLite.Simple.ToField (ToField (..))
-#endif
 
 data AgentSMPServerStats = AgentSMPServerStats
   { sentDirect :: TVar Int, -- successfully sent messages

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

@@ -262,7 +262,7 @@ import Simplex.Messaging.Agent.Stats
 import Simplex.Messaging.Agent.Store
 import Simplex.Messaging.Agent.Store.Common
 import qualified Simplex.Messaging.Agent.Store.DB as DB
-import Simplex.Messaging.Agent.Store.DB (Binary (..), BoolInt (..))
+import Simplex.Messaging.Agent.Store.DB (Binary (..), BoolInt (..), FromField (..), ToField (..))
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Crypto.File (CryptoFile (..), CryptoFileArgs (..))
 import Simplex.Messaging.Crypto.Ratchet (PQEncryption (..), PQSupport (..), RatchetX448, SkippedMsgDiff (..), SkippedMsgKeys)
@@ -281,16 +281,12 @@ import qualified UnliftIO.Exception as E
 import UnliftIO.STM
 #if defined(dbPostgres)
 import Database.PostgreSQL.Simple (Only (..), Query, SqlError, (:.) (..))
-import Database.PostgreSQL.Simple.FromField (FromField (..))
 import Database.PostgreSQL.Simple.Errors (constraintViolation)
 import Database.PostgreSQL.Simple.SqlQQ (sql)
-import Database.PostgreSQL.Simple.ToField (ToField (..))
 #else
 import Database.SQLite.Simple (FromRow (..), Only (..), Query (..), SQLError, ToRow (..), field, (:.) (..))
 import qualified Database.SQLite.Simple as SQL
-import Database.SQLite.Simple.FromField
 import Database.SQLite.Simple.QQ (sql)
-import Database.SQLite.Simple.ToField (ToField (..))
 #endif
 
 checkConstraint :: StoreError -> IO (Either StoreError a) -> IO (Either StoreError a)

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

@@ -3,11 +3,15 @@
 module Simplex.Messaging.Agent.Store.DB
 #if defined(dbPostgres)
   ( module Simplex.Messaging.Agent.Store.Postgres.DB,
+    FromField (..),
+    ToField (..),
   )
   where
 import Simplex.Messaging.Agent.Store.Postgres.DB
 #else
   ( module Simplex.Messaging.Agent.Store.SQLite.DB,
+    FromField (..),
+    ToField (..),
   )
   where
 import Simplex.Messaging.Agent.Store.SQLite.DB

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

@@ -4,6 +4,8 @@ module Simplex.Messaging.Agent.Store.Postgres.DB
   ( BoolInt (..),
     PSQL.Binary (..),
     PSQL.Connection,
+    FromField (..),
+    ToField (..),
     PSQL.connect,
     PSQL.close,
     execute,
@@ -49,15 +51,15 @@ executeMany db q qs = void $ PSQL.executeMany db q qs
 -- used in FileSize
 instance FromField Word32 where
   fromField field dat = do
-    i <- fromField field dat
-    if i >= (0 :: Int64)
+    i :: Int64 <- fromField field dat
+    if i >= 0 && i <= fromIntegral (maxBound :: Word32)
       then pure (fromIntegral i :: Word32)
       else returnError ConversionFailed field "Negative value can't be converted to Word32"
 
 -- used in Version
 instance FromField Word16 where
   fromField field dat = do
-    i <- fromField field dat
-    if i >= (0 :: Int32)
+    i :: Int64 <- fromField field dat
+    if i >= 0 && i <= fromIntegral (maxBound :: Word16)
       then pure (fromIntegral i :: Word16)
       else returnError ConversionFailed field "Negative value can't be converted to Word16"

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

@@ -69,32 +69,33 @@ data DBOpts = DBOpts
   { dbFilePath :: FilePath,
     dbKey :: ScrubbedBytes,
     keepKey :: Bool,
-    vacuum :: Bool
+    vacuum :: Bool,
+    track :: DB.TrackQueries
   }
 
 createDBStore :: DBOpts -> [Migration] -> MigrationConfirmation -> IO (Either MigrationError DBStore)
-createDBStore DBOpts {dbFilePath, dbKey, keepKey, vacuum} migrations confirmMigrations = do
+createDBStore DBOpts {dbFilePath, dbKey, keepKey, track, vacuum} migrations confirmMigrations = do
   let dbDir = takeDirectory dbFilePath
   createDirectoryIfMissing True dbDir
-  st <- connectSQLiteStore dbFilePath dbKey keepKey
+  st <- connectSQLiteStore dbFilePath dbKey keepKey track
   r <- migrateSchema st migrations confirmMigrations vacuum `onException` closeDBStore st
   case r of
     Right () -> pure $ Right st
     Left e -> closeDBStore st $> Left e
 
-connectSQLiteStore :: FilePath -> ScrubbedBytes -> Bool -> IO DBStore
-connectSQLiteStore dbFilePath key keepKey = do
+connectSQLiteStore :: FilePath -> ScrubbedBytes -> Bool -> DB.TrackQueries -> IO DBStore
+connectSQLiteStore dbFilePath key keepKey track = do
   dbNew <- not <$> doesFileExist dbFilePath
-  dbConn <- dbBusyLoop (connectDB dbFilePath key)
+  dbConn <- dbBusyLoop (connectDB dbFilePath key track)
   dbConnection <- newMVar dbConn
   dbKey <- newTVarIO $! storeKey key keepKey
   dbClosed <- newTVarIO False
   dbSem <- newTVarIO 0
   pure DBStore {dbFilePath, dbKey, dbSem, dbConnection, dbNew, dbClosed}
 
-connectDB :: FilePath -> ScrubbedBytes -> IO DB.Connection
-connectDB path key = do
-  db <- DB.open path
+connectDB :: FilePath -> ScrubbedBytes -> DB.TrackQueries -> IO DB.Connection
+connectDB path key track = do
+  db <- DB.open path track
   prepare db `onException` DB.close db
   -- _printPragmas db path
   pure db
@@ -127,12 +128,12 @@ openSQLiteStore_ DBStore {dbConnection, dbFilePath, dbKey, dbClosed} key keepKey
   bracketOnError
     (takeMVar dbConnection)
     (tryPutMVar dbConnection)
-    $ \DB.Connection {slow} -> do
-      DB.Connection {conn} <- connectDB dbFilePath key
+    $ \DB.Connection {slow, track} -> do
+      DB.Connection {conn} <- connectDB dbFilePath key track
       atomically $ do
         writeTVar dbClosed False
         writeTVar dbKey $! storeKey key keepKey
-      putMVar dbConnection DB.Connection {conn, slow}
+      putMVar dbConnection DB.Connection {conn, slow, track}
 
 reopenDBStore :: DBStore -> IO ()
 reopenDBStore st@DBStore {dbKey, dbClosed} =

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

@@ -11,6 +11,9 @@ module Simplex.Messaging.Agent.Store.SQLite.DB
     Binary (..),
     Connection (..),
     SlowQueryStats (..),
+    TrackQueries (..),
+    FromField (..),
+    ToField (..),
     open,
     close,
     execute,
@@ -38,7 +41,7 @@ import Database.SQLite.Simple.ToField (ToField (..))
 import Simplex.Messaging.Parsers (defaultJSON)
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
-import Simplex.Messaging.Util (diffToMilliseconds, tshow)
+import Simplex.Messaging.Util (diffToMicroseconds, tshow)
 
 newtype BoolInt = BI {unBI :: Bool}
   deriving newtype (FromField, ToField)
@@ -48,9 +51,13 @@ newtype Binary = Binary {fromBinary :: ByteString}
 
 data Connection = Connection
   { conn :: SQL.Connection,
+    track :: TrackQueries,
     slow :: TMap Query SlowQueryStats
   }
 
+data TrackQueries = TQAll | TQSlow Int64 | TQOff
+  deriving (Eq)
+
 data SlowQueryStats = SlowQueryStats
   { count :: Int64,
     timeMax :: Int64,
@@ -59,22 +66,29 @@ data SlowQueryStats = SlowQueryStats
   }
   deriving (Show)
 
-timeIt :: TMap Query SlowQueryStats -> Query -> IO a -> IO a
-timeIt slow sql a = do
-  t <- getCurrentTime
-  r <-
-    a `catch` \e -> do
-      atomically $ TM.alter (Just . updateQueryErrors e) sql slow
-      throwIO e
-  t' <- getCurrentTime
-  let diff = diffToMilliseconds $ diffUTCTime t' t
-  when (diff > 1) $ atomically $ TM.alter (updateQueryStats diff) sql slow
-  pure r
+timeIt :: Connection -> Query -> IO a -> IO a
+timeIt Connection {slow, track} sql a
+  | track == TQOff = makeQuery
+  | otherwise = do
+      t <- getCurrentTime
+      r <- makeQuery
+      t' <- getCurrentTime
+      let diff = diffToMicroseconds $ diffUTCTime t' t
+      when (trackQuery diff) $ atomically $ TM.alter (updateQueryStats diff) sql slow
+      pure r
   where
+    makeQuery =
+      a `catch` \e -> do
+        atomically $ TM.alter (Just . updateQueryErrors e) sql slow
+        throwIO e
+    trackQuery diff = case track of
+      TQOff -> False
+      TQSlow t -> diff > t
+      TQAll -> True
     updateQueryErrors :: SomeException -> Maybe SlowQueryStats -> SlowQueryStats
     updateQueryErrors e Nothing = SlowQueryStats 0 0 0 $ M.singleton (tshow e) 1
-    updateQueryErrors e (Just stats@SlowQueryStats {errs}) =
-      stats {errs = M.alter (Just . maybe 1 (+ 1)) (tshow e) errs}
+    updateQueryErrors e (Just st@SlowQueryStats {errs}) =
+      st {errs = M.alter (Just . maybe 1 (+ 1)) (tshow e) errs}
     updateQueryStats :: Int64 -> Maybe SlowQueryStats -> Maybe SlowQueryStats
     updateQueryStats diff Nothing = Just $ SlowQueryStats 1 diff diff M.empty
     updateQueryStats diff (Just SlowQueryStats {count, timeMax, timeAvg, errs}) =
@@ -86,33 +100,33 @@ timeIt slow sql a = do
             errs
           }
 
-open :: String -> IO Connection
-open f = do
+open :: String -> TrackQueries -> IO Connection
+open f track = do
   conn <- SQL.open f
   slow <- TM.emptyIO
-  pure Connection {conn, slow}
+  pure Connection {conn, slow, track}
 
 close :: Connection -> IO ()
 close = SQL.close . conn
 
 execute :: ToRow q => Connection -> Query -> q -> IO ()
-execute Connection {conn, slow} sql = timeIt slow sql . SQL.execute conn sql
+execute c sql = timeIt c sql . SQL.execute (conn c) sql
 {-# INLINE execute #-}
 
 execute_ :: Connection -> Query -> IO ()
-execute_ Connection {conn, slow} sql = timeIt slow sql $ SQL.execute_ conn sql
+execute_ c sql = timeIt c sql $ SQL.execute_ (conn c) sql
 {-# INLINE execute_ #-}
 
 executeMany :: ToRow q => Connection -> Query -> [q] -> IO ()
-executeMany Connection {conn, slow} sql = timeIt slow sql . SQL.executeMany conn sql
+executeMany c sql = timeIt c sql . SQL.executeMany (conn c) sql
 {-# INLINE executeMany #-}
 
 query :: (ToRow q, FromRow r) => Connection -> Query -> q -> IO [r]
-query Connection {conn, slow} sql = timeIt slow sql . SQL.query conn sql
+query c sql = timeIt c sql . SQL.query (conn c) sql
 {-# INLINE query #-}
 
 query_ :: FromRow r => Connection -> Query -> IO [r]
-query_ Connection {conn, slow} sql = timeIt slow sql $ SQL.query_ conn sql
+query_ c sql = timeIt c sql $ SQL.query_ (conn c) sql
 {-# INLINE query_ #-}
 
 $(J.deriveJSON defaultJSON ''SlowQueryStats)

src/Simplex/Messaging/Crypto.hs

@@ -238,18 +238,11 @@ import Data.X509
 import Data.X509.Validation (Fingerprint (..), getFingerprint)
 import GHC.TypeLits (ErrorMessage (..), KnownNat, Nat, TypeError, natVal, type (+))
 import Network.Transport.Internal (decodeWord16, encodeWord16)
-import Simplex.Messaging.Agent.Store.DB (Binary (..))
+import Simplex.Messaging.Agent.Store.DB (Binary (..), FromField (..), ToField (..))
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (blobFieldDecoder, parseAll, parseString)
 import Simplex.Messaging.Util ((<$?>))
-#if defined(dbPostgres)
-import Database.PostgreSQL.Simple.FromField (FromField (..))
-import Database.PostgreSQL.Simple.ToField (ToField (..))
-#else
-import Database.SQLite.Simple.FromField (FromField (..))
-import Database.SQLite.Simple.ToField (ToField (..))
-#endif
 
 -- | Cryptographic algorithms.
 data Algorithm = Ed25519 | Ed448 | X25519 | X448

src/Simplex/Messaging/Crypto/Ratchet.hs

@@ -111,7 +111,7 @@ import Data.Type.Equality
 import Data.Typeable (Typeable)
 import Data.Word (Word16, Word32)
 import Simplex.Messaging.Agent.QueryString
-import Simplex.Messaging.Agent.Store.DB (Binary (..), BoolInt (..))
+import Simplex.Messaging.Agent.Store.DB (Binary (..), BoolInt (..), FromField (..), ToField (..))
 import Simplex.Messaging.Crypto
 import Simplex.Messaging.Crypto.SNTRUP761.Bindings
 import Simplex.Messaging.Encoding
@@ -121,13 +121,6 @@ import Simplex.Messaging.Util (($>>=), (<$?>))
 import Simplex.Messaging.Version
 import Simplex.Messaging.Version.Internal
 import UnliftIO.STM
-#if defined(dbPostgres)
-import Database.PostgreSQL.Simple.FromField (FromField (..))
-import Database.PostgreSQL.Simple.ToField (ToField (..))
-#else
-import Database.SQLite.Simple.FromField (FromField (..))
-import Database.SQLite.Simple.ToField (ToField (..))
-#endif
 
 -- e2e encryption headers version history:
 -- 1 - binary protocol encoding (1/1/2022)

src/Simplex/Messaging/Crypto/SNTRUP761/Bindings.hs

@@ -11,18 +11,12 @@ import Data.ByteArray (ScrubbedBytes)
 import qualified Data.ByteArray as BA
 import Data.ByteString (ByteString)
 import Foreign (nullPtr)
+import Simplex.Messaging.Agent.Store.DB (FromField (..), ToField (..))
 import Simplex.Messaging.Crypto.SNTRUP761.Bindings.Defines
 import Simplex.Messaging.Crypto.SNTRUP761.Bindings.FFI
 import Simplex.Messaging.Crypto.SNTRUP761.Bindings.RNG (withDRG)
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
-#if defined(dbPostgres)
-import Database.PostgreSQL.Simple.FromField
-import Database.PostgreSQL.Simple.ToField
-#else
-import Database.SQLite.Simple.FromField
-import Database.SQLite.Simple.ToField
-#endif
 
 newtype KEMPublicKey = KEMPublicKey ByteString
   deriving (Eq, Show)

src/Simplex/Messaging/Notifications/Protocol.hs

@@ -1,4 +1,3 @@
-{-# LANGUAGE CPP #-}
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE LambdaCase #-}
@@ -29,6 +28,7 @@ import Data.Time.Clock.System
 import Data.Type.Equality
 import Data.Word (Word16)
 import Simplex.Messaging.Agent.Protocol (updateSMPServerHosts)
+import Simplex.Messaging.Agent.Store.DB (FromField (..), ToField (..))
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
@@ -36,13 +36,6 @@ import Simplex.Messaging.Notifications.Transport (NTFVersion, ntfClientHandshake
 import Simplex.Messaging.Parsers (fromTextField_)
 import Simplex.Messaging.Protocol hiding (Command (..), CommandTag (..))
 import Simplex.Messaging.Util (eitherToMaybe, (<$?>))
-#if defined(dbPostgres)
-import Database.PostgreSQL.Simple.FromField (FromField (..))
-import Database.PostgreSQL.Simple.ToField (ToField (..))
-#else
-import Database.SQLite.Simple.FromField (FromField (..))
-import Database.SQLite.Simple.ToField (ToField (..))
-#endif
 
 data NtfEntity = Token | Subscription
   deriving (Show)

src/Simplex/Messaging/Notifications/Types.hs

@@ -1,4 +1,3 @@
-{-# LANGUAGE CPP #-}
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE LambdaCase #-}
@@ -11,19 +10,12 @@ import qualified Data.Attoparsec.ByteString.Char8 as A
 import Data.Text.Encoding (decodeLatin1, encodeUtf8)
 import Data.Time (UTCTime)
 import Simplex.Messaging.Agent.Protocol (ConnId, NotificationsMode (..), UserId)
-import Simplex.Messaging.Agent.Store.DB (Binary (..))
+import Simplex.Messaging.Agent.Store.DB (Binary (..), FromField (..), ToField (..))
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Notifications.Protocol
 import Simplex.Messaging.Parsers (blobFieldDecoder, fromTextField_)
 import Simplex.Messaging.Protocol (NotifierId, NtfServer, SMPServer)
-#if defined(dbPostgres)
-import Database.PostgreSQL.Simple.FromField (FromField (..))
-import Database.PostgreSQL.Simple.ToField (ToField (..))
-#else
-import Database.SQLite.Simple.FromField (FromField (..))
-import Database.SQLite.Simple.ToField (ToField (..))
-#endif
 
 data NtfTknAction
   = NTARegister

src/Simplex/Messaging/Server.hs

@@ -1150,9 +1150,10 @@ client
   ms
   clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId, procThreads} = do
     labelMyThread . B.unpack $ "client $" <> encode sessionId <> " commands"
+    let THandleParams {thVersion} = thParams'
     forever $
       atomically (readTBQueue rcvQ)
-        >>= mapM processCommand
+        >>= mapM (processCommand thVersion)
         >>= mapM_ reply . L.nonEmpty . catMaybes . L.toList
   where
     reply :: MonadIO m => NonEmpty (Transmission BrokerMsg) -> m ()
@@ -1237,8 +1238,8 @@ client
     mkIncProxyStats ps psOwn own sel = do
       incStat $ sel ps
       when own $ incStat $ sel psOwn
-    processCommand :: (Maybe (StoreQueue s, QueueRec), Transmission Cmd) -> M (Maybe (Transmission BrokerMsg))
-    processCommand (q_, (corrId, entId, cmd)) = case cmd of
+    processCommand :: VersionSMP -> (Maybe (StoreQueue s, QueueRec), Transmission Cmd) -> M (Maybe (Transmission BrokerMsg))
+    processCommand clntVersion (q_, (corrId, entId, cmd)) = case cmd of
       Cmd SProxiedClient command -> processProxiedCmd (corrId, entId, command)
       Cmd SSender command -> Just <$> case command of
         SKEY sKey ->
@@ -1499,7 +1500,7 @@ client
 
         sendMessage :: MsgFlags -> MsgBody -> StoreQueue s -> QueueRec -> M (Transmission BrokerMsg)
         sendMessage msgFlags msgBody q qr
-          | B.length msgBody > maxMessageLength thVersion = do
+          | B.length msgBody > maxMessageLength clntVersion = do
               stats <- asks serverStats
               incStat $ msgSentLarge stats
               pure $ err LARGE_MSG
@@ -1538,7 +1539,6 @@ client
                           liftIO $ updatePeriodStats (activeQueues stats) (recipientId' q)
                           pure ok
           where
-            THandleParams {thVersion} = thParams'
             mkMessage :: MsgId -> C.MaxLenBS MaxMessageLen -> IO Message
             mkMessage msgId body = do
               msgTs <- getSystemTime
@@ -1647,7 +1647,7 @@ client
               Left r -> pure r
               -- rejectOrVerify filters allowed commands, no need to repeat it here.
               -- INTERNAL is used because processCommand never returns Nothing for sender commands (could be extracted for better types).
-              Right t''@(_, (corrId', entId', _)) -> fromMaybe (corrId', entId', ERR INTERNAL) <$> lift (processCommand t'')
+              Right t''@(_, (corrId', entId', _)) -> fromMaybe (corrId', entId', ERR INTERNAL) <$> lift (processCommand fwdVersion t'')
           -- encode response
           r' <- case batchTransmissions (batch clntTHParams) (blockSize clntTHParams) [Right (Nothing, encodeTransmission clntTHParams r)] of
             [] -> throwE INTERNAL -- at least 1 item is guaranteed from NonEmpty/Right

src/Simplex/RemoteControl/Types.hs

@@ -1,4 +1,3 @@
-{-# LANGUAGE CPP #-}
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DeriveAnyClass #-}
 {-# LANGUAGE DuplicateRecordFields #-}

tests/AgentTests/FunctionalAPITests.hs

@@ -2055,8 +2055,8 @@ testAsyncCommands sqSecured alice bob baseId =
     ackMessageAsync alice "7" bobId (baseId + 4) Nothing
     get alice =##> \case ("7", _, OK) -> True; _ -> False
     deleteConnectionAsync alice False bobId
-    get alice =##> \case ("", c, DEL_RCVQ _ _ Nothing) -> c == bobId; _ -> False
-    get alice =##> \case ("", c, DEL_CONN) -> c == bobId; _ -> False
+    get alice =##> \case ("", "", DEL_RCVQS [(c, _, _, Nothing)]) -> c == bobId; _ -> False
+    get alice =##> \case ("", "", DEL_CONNS [c]) -> c == bobId; _ -> False
     liftIO $ noMessages alice "nothing else should be delivered to alice"
   where
     msgId = subtract baseId
@@ -2123,12 +2123,9 @@ testDeleteConnectionAsync t =
     runRight_ $ do
       deleteConnectionsAsync a False connIds
       nGet a =##> \case ("", "", DOWN {}) -> True; _ -> False
-      get a =##> \case ("", c, DEL_RCVQ _ _ (Just (BROKER _ e))) -> c `elem` connIds && (e == TIMEOUT || e == NETWORK); _ -> False
-      get a =##> \case ("", c, DEL_RCVQ _ _ (Just (BROKER _ e))) -> c `elem` connIds && (e == TIMEOUT || e == NETWORK); _ -> False
-      get a =##> \case ("", c, DEL_RCVQ _ _ (Just (BROKER _ e))) -> c `elem` connIds && (e == TIMEOUT || e == NETWORK); _ -> False
-      get a =##> \case ("", c, DEL_CONN) -> c `elem` connIds; _ -> False
-      get a =##> \case ("", c, DEL_CONN) -> c `elem` connIds; _ -> False
-      get a =##> \case ("", c, DEL_CONN) -> c `elem` connIds; _ -> False
+      let delOk = \case (c, _, _, Just (BROKER _ e)) -> c `elem` connIds && (e == TIMEOUT || e == NETWORK); _ -> False
+      get a =##> \case ("", "", DEL_RCVQS rs) -> length rs == 3 && all delOk rs; _ -> False
+      get a =##> \case ("", "", DEL_CONNS cs) -> length cs == 3 && all (`elem` connIds) cs; _ -> False
       liftIO $ noMessages a "nothing else should be delivered to alice"
 
 testWaitDeliveryNoPending :: ATransport -> IO ()
@@ -2147,8 +2144,8 @@ testWaitDeliveryNoPending t = withAgentClients2 $ \alice bob ->
     ackMessage alice bobId (baseId + 2) Nothing
 
     deleteConnectionsAsync alice True [bobId]
-    get alice =##> \case ("", cId, DEL_RCVQ _ _ Nothing) -> cId == bobId; _ -> False
-    get alice =##> \case ("", cId, DEL_CONN) -> cId == bobId; _ -> False
+    get alice =##> \case ("", "", DEL_RCVQS [(cId, _, _, Nothing)]) -> cId == bobId; _ -> False
+    get alice =##> \case ("", "", DEL_CONNS [cId]) -> cId == bobId; _ -> False
 
     3 <- msgId <$> sendMessage bob aliceId SMP.noMsgFlags "message 2"
     get bob =##> \case ("", cId, MERR mId (SMP _ AUTH)) -> cId == aliceId && mId == (baseId + 3); _ -> False
@@ -2184,14 +2181,14 @@ testWaitDelivery t =
         3 <- msgId <$> sendMessage alice bobId SMP.noMsgFlags "how are you?"
         4 <- msgId <$> sendMessage alice bobId SMP.noMsgFlags "message 1"
         deleteConnectionsAsync alice True [bobId]
-        get alice =##> \case ("", cId, DEL_RCVQ _ _ (Just (BROKER _ e))) -> cId == bobId && (e == TIMEOUT || e == NETWORK); _ -> False
+        get alice =##> \case ("", "", DEL_RCVQS [(cId, _, _, Just (BROKER _ e))]) -> cId == bobId && (e == TIMEOUT || e == NETWORK); _ -> False
         liftIO $ noMessages alice "nothing else should be delivered to alice"
         liftIO $ noMessages bob "nothing else should be delivered to bob"
 
       withSmpServerStoreLogOn t testPort $ \_ -> runRight_ $ do
         get alice ##> ("", bobId, SENT $ baseId + 3)
         get alice ##> ("", bobId, SENT $ baseId + 4)
-        get alice =##> \case ("", cId, DEL_CONN) -> cId == bobId; _ -> False
+        get alice =##> \case ("", "", DEL_CONNS [cId]) -> cId == bobId; _ -> False
 
         liftIO $
           getInAnyOrder
@@ -2231,8 +2228,8 @@ testWaitDeliveryAUTHErr t =
         ackMessage alice bobId (baseId + 2) Nothing
 
         deleteConnectionsAsync bob False [aliceId]
-        get bob =##> \case ("", cId, DEL_RCVQ _ _ Nothing) -> cId == aliceId; _ -> False
-        get bob =##> \case ("", cId, DEL_CONN) -> cId == aliceId; _ -> False
+        get bob =##> \case ("", "", DEL_RCVQS [(cId, _, _, Nothing)]) -> cId == aliceId; _ -> False
+        get bob =##> \case ("", "", DEL_CONNS [cId]) -> cId == aliceId; _ -> False
 
         pure (aliceId, bobId)
 
@@ -2241,14 +2238,14 @@ testWaitDeliveryAUTHErr t =
         3 <- msgId <$> sendMessage alice bobId SMP.noMsgFlags "how are you?"
         4 <- msgId <$> sendMessage alice bobId SMP.noMsgFlags "message 1"
         deleteConnectionsAsync alice True [bobId]
-        get alice =##> \case ("", cId, DEL_RCVQ _ _ (Just (BROKER _ e))) -> cId == bobId && (e == TIMEOUT || e == NETWORK); _ -> False
+        get alice =##> \case ("", "", DEL_RCVQS [(cId, _, _, Just (BROKER _ e))]) -> cId == bobId && (e == TIMEOUT || e == NETWORK); _ -> False
         liftIO $ noMessages alice "nothing else should be delivered to alice"
         liftIO $ noMessages bob "nothing else should be delivered to bob"
 
       withSmpServerStoreLogOn t testPort $ \_ -> do
         get alice =##> \case ("", cId, MERR mId (SMP _ AUTH)) -> cId == bobId && mId == (baseId + 3); _ -> False
         get alice =##> \case ("", cId, MERR mId (SMP _ AUTH)) -> cId == bobId && mId == (baseId + 4); _ -> False
-        get alice =##> \case ("", cId, DEL_CONN) -> cId == bobId; _ -> False
+        get alice =##> \case ("", "", DEL_CONNS [cId]) -> cId == bobId; _ -> False
 
         liftIO $ noMessages alice "nothing else should be delivered to alice"
         liftIO $ noMessages bob "nothing else should be delivered to bob"
@@ -2281,8 +2278,8 @@ testWaitDeliveryTimeout t =
         3 <- msgId <$> sendMessage alice bobId SMP.noMsgFlags "how are you?"
         4 <- msgId <$> sendMessage alice bobId SMP.noMsgFlags "message 1"
         deleteConnectionsAsync alice True [bobId]
-        get alice =##> \case ("", cId, DEL_RCVQ _ _ (Just (BROKER _ e))) -> cId == bobId && (e == TIMEOUT || e == NETWORK); _ -> False
-        get alice =##> \case ("", cId, DEL_CONN) -> cId == bobId; _ -> False
+        get alice =##> \case ("", "", DEL_RCVQS [(cId, _, _, Just (BROKER _ e))]) -> cId == bobId && (e == TIMEOUT || e == NETWORK); _ -> False
+        get alice =##> \case ("", "", DEL_CONNS [cId]) -> cId == bobId; _ -> False
         liftIO $ noMessages alice "nothing else should be delivered to alice"
         liftIO $ noMessages bob "nothing else should be delivered to bob"
 
@@ -2321,8 +2318,8 @@ testWaitDeliveryTimeout2 t =
         3 <- msgId <$> sendMessage alice bobId SMP.noMsgFlags "how are you?"
         4 <- msgId <$> sendMessage alice bobId SMP.noMsgFlags "message 1"
         deleteConnectionsAsync alice True [bobId]
-        get alice =##> \case ("", cId, DEL_RCVQ _ _ (Just (BROKER _ e))) -> cId == bobId && (e == TIMEOUT || e == NETWORK); _ -> False
-        get alice =##> \case ("", cId, DEL_CONN) -> cId == bobId; _ -> False
+        get alice =##> \case ("", "", DEL_RCVQS [(cId, _, _, Just (BROKER _ e))]) -> cId == bobId && (e == TIMEOUT || e == NETWORK); _ -> False
+        get alice =##> \case ("", "", DEL_CONNS [cId]) -> cId == bobId; _ -> False
         liftIO $ noMessages alice "nothing else should be delivered to alice"
         liftIO $ noMessages bob "nothing else should be delivered to bob"
 
@@ -2430,8 +2427,8 @@ testUsers =
     (aId', bId') <- makeConnectionForUsers a auId b 1
     exchangeGreetings a bId' b aId'
     deleteUser a auId True
-    get a =##> \case ("", c, DEL_RCVQ _ _ Nothing) -> c == bId'; _ -> False
-    get a =##> \case ("", c, DEL_CONN) -> c == bId'; _ -> False
+    get a =##> \case ("", "", DEL_RCVQS [(c, _, _, Nothing)]) -> c == bId'; _ -> False
+    get a =##> \case ("", "", DEL_CONNS [c]) -> c == bId'; _ -> False
     nGet a =##> \case ("", "", DEL_USER u) -> u == auId; _ -> False
     exchangeGreetingsMsgId 4 a bId b aId
     liftIO $ noMessages a "nothing else should be delivered to alice"
@@ -2462,8 +2459,8 @@ testUsersNoServer t = withAgentClientsCfg2 aCfg agentCfg $ \a b -> do
   nGet b =##> \case ("", "", DOWN _ cs) -> length cs == 2; _ -> False
   runRight_ $ do
     deleteUser a auId True
-    get a =##> \case ("", c, DEL_RCVQ _ _ (Just (BROKER _ e))) -> c == bId' && (e == TIMEOUT || e == NETWORK); _ -> False
-    get a =##> \case ("", c, DEL_CONN) -> c == bId'; _ -> False
+    get a =##> \case ("", "", DEL_RCVQS [(c, _, _, Just (BROKER _ e))]) -> c == bId' && (e == TIMEOUT || e == NETWORK); _ -> False
+    get a =##> \case ("", "", DEL_CONNS [c]) -> c == bId'; _ -> False
     nGet a =##> \case ("", "", DEL_USER u) -> u == auId; _ -> False
     liftIO $ noMessages a "nothing else should be delivered to alice"
   withSmpServerStoreLogOn t testPort $ \_ -> runRight_ $ do
@@ -2581,9 +2578,8 @@ testSwitchDelete servers =
     liftIO $ rcvSwchStatuses' stats `shouldMatchList` [Just RSSwitchStarted]
     phaseRcv a bId SPStarted [Just RSSendingQADD, Nothing]
     deleteConnectionAsync a False bId
-    get a =##> \case ("", c, DEL_RCVQ _ _ Nothing) -> c == bId; _ -> False
-    get a =##> \case ("", c, DEL_RCVQ _ _ Nothing) -> c == bId; _ -> False
-    get a =##> \case ("", c, DEL_CONN) -> c == bId; _ -> False
+    get a =##> \case ("", "", DEL_RCVQS [(c, _, _, Nothing), (c', _, _, Nothing)]) -> c == bId && c' == bId; _ -> False
+    get a =##> \case ("", "", DEL_CONNS [c]) -> c == bId; _ -> False
     liftIO $ noMessages a "nothing else should be delivered to alice"
 
 testAbortSwitchStarted :: HasCallStack => InitialAgentServers -> IO ()
@@ -3104,7 +3100,7 @@ insertUser :: DBStore -> IO ()
 insertUser st = withTransaction st (`DB.execute_` "INSERT INTO users DEFAULT VALUES")
 #else
 createStore :: String -> IO (Either MigrationError DBStore)
-createStore dbPath = createAgentStore (DBOpts dbPath "" False True) MCError
+createStore dbPath = createAgentStore (DBOpts dbPath "" False True DB.TQOff) MCError
 
 insertUser :: DBStore -> IO ()
 insertUser st = withTransaction st (`DB.execute_` "INSERT INTO users (user_id) VALUES (1)")

tests/AgentTests/MigrationTests.hs

@@ -228,7 +228,8 @@ createStore randSuffix migrations confirmMigrations = do
           dbFilePath = testDB randSuffix,
           dbKey = "",
           keepKey = False,
-          vacuum = True
+          vacuum = True,
+          track = DB.TQOff
         }
   createDBStore dbOpts migrations confirmMigrations
 

tests/AgentTests/SQLiteTests.hs

@@ -70,7 +70,7 @@ withStore2 = before connect2 . after (removeStore . fst)
     connect2 :: IO (DBStore, DBStore)
     connect2 = do
       s1@DBStore {dbFilePath} <- createStore'
-      s2 <- connectSQLiteStore dbFilePath "" False
+      s2 <- connectSQLiteStore dbFilePath "" False DB.TQOff
       pure (s1, s2)
 
 createStore' :: IO DBStore
@@ -81,7 +81,7 @@ createEncryptedStore key keepKey = do
   -- Randomize DB file name to avoid SQLite IO errors supposedly caused by asynchronous
   -- IO operations on multiple similarly named files; error seems to be environment specific
   r <- randomIO :: IO Word32
-  Right st <- createDBStore (DBOpts (testDB <> show r) key keepKey True) Migrations.app MCError
+  Right st <- createDBStore (DBOpts (testDB <> show r) key keepKey True DB.TQOff) Migrations.app MCError
   withTransaction' st (`SQL.execute_` "INSERT INTO users (user_id) VALUES (1);")
   pure st
 

tests/AgentTests/SchemaDump.hs

@@ -12,6 +12,7 @@ import Database.SQLite.Simple (Only (..))
 import qualified Database.SQLite.Simple as SQL
 import Simplex.Messaging.Agent.Store.SQLite
 import Simplex.Messaging.Agent.Store.SQLite.Common (withTransaction')
+import Simplex.Messaging.Agent.Store.SQLite.DB (TrackQueries (..))
 import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations
 import Simplex.Messaging.Agent.Store.Shared (Migration (..), MigrationConfirmation (..), MigrationsToRun (..), toDownMigration)
 import Simplex.Messaging.Util (ifM)
@@ -49,7 +50,7 @@ testVerifySchemaDump :: IO ()
 testVerifySchemaDump = do
   savedSchema <- ifM (doesFileExist appSchema) (readFile appSchema) (pure "")
   savedSchema `deepseq` pure ()
-  void $ createDBStore (DBOpts testDB "" False True) Migrations.app MCConsole
+  void $ createDBStore (DBOpts testDB "" False True TQOff) Migrations.app MCConsole
   getSchema testDB appSchema `shouldReturn` savedSchema
   removeFile testDB
 
@@ -57,7 +58,7 @@ testVerifyLintFKeyIndexes :: IO ()
 testVerifyLintFKeyIndexes = do
   savedLint <- ifM (doesFileExist appLint) (readFile appLint) (pure "")
   savedLint `deepseq` pure ()
-  void $ createDBStore (DBOpts testDB "" False True) Migrations.app MCConsole
+  void $ createDBStore (DBOpts testDB "" False True TQOff) Migrations.app MCConsole
   getLintFKeyIndexes testDB "tests/tmp/agent_lint.sql" `shouldReturn` savedLint
   removeFile testDB
 
@@ -70,7 +71,7 @@ withTmpFiles =
 testSchemaMigrations :: IO ()
 testSchemaMigrations = do
   let noDownMigrations = dropWhileEnd (\Migration {down} -> isJust down) Migrations.app
-  Right st <- createDBStore (DBOpts testDB "" False True) noDownMigrations MCError
+  Right st <- createDBStore (DBOpts testDB "" False True TQOff) noDownMigrations MCError
   mapM_ (testDownMigration st) $ drop (length noDownMigrations) Migrations.app
   closeDBStore st
   removeFile testDB
@@ -93,7 +94,7 @@ testSchemaMigrations = do
 
 testUsersMigrationNew :: IO ()
 testUsersMigrationNew = do
-  Right st <- createDBStore (DBOpts testDB "" False True) Migrations.app MCError
+  Right st <- createDBStore (DBOpts testDB "" False True TQOff) Migrations.app MCError
   withTransaction' st (`SQL.query_` "SELECT user_id FROM users;")
     `shouldReturn` ([] :: [Only Int])
   closeDBStore st
@@ -101,11 +102,11 @@ testUsersMigrationNew = do
 testUsersMigrationOld :: IO ()
 testUsersMigrationOld = do
   let beforeUsers = takeWhile (("m20230110_users" /=) . name) Migrations.app
-  Right st <- createDBStore (DBOpts testDB "" False True) beforeUsers MCError
+  Right st <- createDBStore (DBOpts testDB "" False True TQOff) beforeUsers MCError
   withTransaction' st (`SQL.query_` "SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'users';")
     `shouldReturn` ([] :: [Only String])
   closeDBStore st
-  Right st' <- createDBStore (DBOpts testDB "" False True) Migrations.app MCYesUp
+  Right st' <- createDBStore (DBOpts testDB "" False True TQOff) Migrations.app MCYesUp
   withTransaction' st' (`SQL.query_` "SELECT user_id FROM users;")
     `shouldReturn` ([Only (1 :: Int)])
   closeDBStore st'