Merge branch 'master' into ep/test-coverage

add more message delivery tests (#763 )
Docker entrypoint fixes (#753 )
2026-08-31 00:58:22 +00:00 · 2023-06-02 18:05:07 +04:00 · 2023-06-02 18:00:24 +04:00 · 2023-05-29 14:12:23 +03:00 · 2023-05-29 09:36:37 +01:00 · 2023-05-27 18:28:14 +01:00
43 changed files with 1185 additions and 338 deletions
@@ -1 +1,4 @@
-* @epoberezkin @efim-poberezkin
+* @epoberezkin @spaced4ndy
+/Dockerfile @shumvgolove
+/scripts/docker/ @shumvgolove
+/scripts/main/ @shumvgolove
@@ -0,0 +1,50 @@
+name: Build and push Docker image to Docker Hub
+
+on:
+  push:
+    tags:
+      - "v*"
+
+jobs:
+  build-and-push:
+    name: Build and push Docker image
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+          - app: smp-server
+            app_port: 5223
+          - app: xftp-server
+            app_port: 443 
+    steps:
+      - name: Clone project
+        uses: actions/checkout@v3
+
+      - name: Log in to Docker Hub
+        uses: docker/login-action@v2
+        with:
+          username: ${{ secrets.DOCKERHUB_USERNAME }}
+          password: ${{ secrets.DOCKERHUB_PASSWORD }}
+
+      - name: Extract metadata for Docker image
+        id: meta
+        uses: docker/metadata-action@v4
+        with:
+          images: ${{ secrets.DOCKERHUB_USERNAME }}/${{ matrix.app }}
+          flavor: |
+            latest=auto
+          tags: |
+            type=semver,pattern=v{{version}}
+            type=semver,pattern=v{{major}}.{{minor}}
+            type=semver,pattern=v{{major}}
+
+      - name: Build and push Docker image
+        uses: docker/build-push-action@v4
+        with:
+          push: true
+          build-args: |
+            APP=${{ matrix.app }}
+            APP_PORT=${{ matrix.app_port }}
+          tags: ${{ steps.meta.outputs.tags }}
+          labels: ${{ steps.meta.outputs.labels }}
@@ -7,3 +7,7 @@ dist-newstyle/

 cabal.project.local
 cabal.project.local~
+
+.hpc/
+*.tix
+.coverage
@@ -0,0 +1,62 @@
+ARG TAG=22.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
+
+# Specify bootstrap Haskell versions
+ENV BOOTSTRAP_HASKELL_GHC_VERSION=8.10.7
+ENV BOOTSTRAP_HASKELL_CABAL_VERSION=3.6.2.0
+
+# Install ghcup
+RUN curl --proto '=https' --tlsv1.2 -sSf https://get-ghcup.haskell.org | BOOTSTRAP_HASKELL_NONINTERACTIVE=1 sh
+
+# Adjust PATH
+ENV PATH="/root/.cabal/bin:/root/.ghcup/bin:$PATH"
+
+# Set both as default
+RUN ghcup set ghc "${BOOTSTRAP_HASKELL_GHC_VERSION}" && \
+    ghcup set cabal
+
+COPY . /project
+WORKDIR /project
+
+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
+
+# Compile app
+RUN cabal update
+RUN cabal build exe:$APP
+
+# 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) && \
+    mv "$bin" ./ && \
+    strip ./"$APP" &&\
+    mv /project/scripts/docker/entrypoint-"$APP" ./entrypoint
+
+### Final stage
+FROM ubuntu:${TAG}
+
+# Install OpenSSL dependency
+RUN apt-get update && apt-get install -y openssl libnuma-dev
+
+# Copy compiled app from build stage
+COPY --from=build /final /usr/local/bin/
+
+# Open app listening port
+ARG APP_PORT
+EXPOSE $APP_PORT
+
+# simplexmq requires using SIGINT to correctly preserve undelivered messages and restore them on restart
+STOPSIGNAL SIGINT
+
+# Finally, execute helper script
+ENTRYPOINT [ "/usr/local/bin/entrypoint" ]
@@ -90,11 +90,11 @@ You can either run your own SMP server locally or deploy using [Linode StackScri

 It's the easiest to try SMP agent via a prototype [simplex-chat](https://github.com/simplex-chat/simplex-chat) terminal UI.

-## Deploy SMP server on Linux
+## Deploy SMP/XFTP servers on Linux

-You can run your SMP server as a Linux process, optionally using a service manager for booting and restarts.
+You can run your SMP/XFTP server as a Linux process, optionally using a service manager for booting and restarts.

-Notice that `smp-server` requires `openssl` as run-time dependency (it is used to generate server certificates during initialization). Install it with your packet manager:
+Notice that `smp-server` and `xftp-server` requires `openssl` as run-time dependency (it is used to generate server certificates during initialization). Install it with your packet manager:

 ```sh
 # For Ubuntu
@@ -105,28 +105,53 @@ apt update && apt install openssl

 #### Using Docker

-On Linux, you can deploy smp server using Docker. This will download image from [Docker Hub](https://hub.docker.com/r/simplexchat/simplexmq).
+On Linux, you can deploy smp and xftp server using Docker. This will download image from [Docker Hub](https://hub.docker.com/r/simplexchat).

-1. Create `config` and `logs` directories:
+1. Create directories for persistent Docker configuration:

   ```sh
-   mkdir -p ~/simplex/{config,logs}
+   mkdir -p $HOME/simplex/{xftp,smp}/{config,logs} && mkdir -p $HOME/simplex/xftp/files
   ```

-2. Run your Docker container. You must change **your_ip_or_domain**. `-e "pass=password"` is optional variable to password-protect your `smp` server:
-   ```sh
-   docker run -d \
-       -e "addr=your_ip_or_domain" \
-       -e "pass=password" \
-       -p 5223:5223 \
-       -v $HOME/simplex/config:/etc/opt/simplex:z \
-       -v $HOME/simplex/logs:/var/opt/simplex:z \
-       simplexchat/simplexmq:latest
-   ```
+2. Run your Docker container.

-#### Ubuntu
+   - `smp-server`
+   
+     You must change **your_ip_or_domain**. `-e "pass=password"` is optional variable to password-protect your `smp` server:
+     ```sh
+     docker run -d \
+         -e "ADDR=your_ip_or_domain" \
+         -e "PASS=password" \
+         -p 5223:5223 \
+         -v $HOME/simplex/smp/config:/etc/opt/simplex:z \
+         -v $HOME/simplex/smp/logs:/var/opt/simplex:z \
+         simplexchat/smp-server:latest
+     ```

-For Ubuntu you can download a binary from [the latest release](https://github.com/simplex-chat/simplexmq/releases).
+   - `xftp-server`
+   
+     You must change **your_ip_or_domain** and **maximum_storage**.
+     ```sh
+     docker run -d \
+         -e "ADDR=your_ip_or_domain" \
+         -e "QUOTA=maximum_storage" \
+         -p 443:443 \
+         -v $HOME/simplex/xftp/config:/etc/opt/simplex-xftp:z \
+         -v $HOME/simplex/xftp/logs:/var/opt/simplex-xftp:z \
+         -v $HOME/simplex/xftp/files:/srv/xftp:z \
+         simplexchat/xftp-server:latest
+     ```
+
+#### Using installation script
+
+**Please note** that currently, only Ubuntu distribution is supported.
+
+You can install and setup servers automatically using our script:
+
+```sh
+curl --proto '=https' --tlsv1.2 -sSf https://raw.githubusercontent.com/simplex-chat/simplexmq/stable/install.sh -o simplex-server-install.sh \
+&& if echo '1268d605e90bca1a8c7ef476038a8bd9aa9b1a28f79ea0a1485669ccf8fc23cd simplex-server-install.sh' | sha256sum -c; then chmod +x ./simplex-server-install.sh && ./simplex-server-install.sh; rm /simplex-server-install.sh; else echo "SHA-256 checksum is incorrect!" && rm ./simplex-server-install.sh; fi
+```

 ### Build from source

@@ -136,31 +161,50 @@ For Ubuntu you can download a binary from [the latest release](https://github.co

 On Linux, you can build smp server using Docker.

-1. Build your `smp-server` image:
+1. Build your images:

   ```sh
   git clone https://github.com/simplex-chat/simplexmq
   cd simplexmq
   git checkout stable
-   DOCKER_BUILDKIT=1 docker build -t smp-server -f ./build.Dockerfile .
+   DOCKER_BUILDKIT=1 docker build -t local/smp-server --build-arg APP="smp-server" --build-arg APP_PORT="5223" . # For xmp-server
+   DOCKER_BUILDKIT=1 docker build -t local/xftp-server --build-arg APP="xftp-server" --build-arg APP_PORT="443" . # For xftp-server
   ```

-2. Create `config` and `logs` directories:
+2. Create directories for persistent Docker configuration:

   ```sh
-   mkdir -p ~/simplex/{config,logs}
+   mkdir -p $HOME/simplex/{xftp,smp}/{config,logs} && mkdir -p $HOME/simplex/xftp/files
   ```

-3. Run your Docker container. You must change **your_ip_or_domain**. `-e pass="password"` is optional variable to password-protect your `smp` server::
-   ```sh
-   docker run -d \
-       -e "addr=your_ip_or_domain" \
-       -e "pass=password" \
-       -p 5223:5223 \
-       -v $HOME/simplex/config:/etc/opt/simplex:z \
-       -v $HOME/simplex/logs:/var/opt/simplex:z \
-       smp-server
-   ```
+3. Run your Docker container.
+
+   - `smp-server`
+   
+     You must change **your_ip_or_domain**. `-e "pass=password"` is optional variable to password-protect your `smp` server:
+     ```sh
+     docker run -d \
+         -e "ADDR=your_ip_or_domain" \
+         -e "PASS=password" \
+         -p 5223:5223 \
+         -v $HOME/simplex/smp/config:/etc/opt/simplex:z \
+         -v $HOME/simplex/smp/logs:/var/opt/simplex:z \
+         simplexchat/smp-server:latest
+     ```
+
+   - `xftp-server`
+   
+     You must change **your_ip_or_domain** and **maximum_storage**.
+     ```sh
+     docker run -d \
+         -e "ADDR=your_ip_or_domain" \
+         -e "QUOTA=maximum_storage" \
+         -p 443:443 \
+         -v $HOME/simplex/xftp/config:/etc/opt/simplex-xftp:z \
+         -v $HOME/simplex/xftp/logs:/var/opt/simplex-xftp:z \
+         -v $HOME/simplex/xftp/files:/srv/xftp:z \
+         simplexchat/xftp-server:latest
+     ```

 #### Using your distribution

@@ -1,46 +0,0 @@
-FROM ubuntu:focal AS final
-FROM ubuntu:focal AS build
-
-### Build stage
-
-# Install curl and git and smp-related dependencies
-RUN apt-get update && apt-get install -y curl git build-essential libgmp3-dev zlib1g-dev llvm llvm-dev libnuma-dev
-
-# Install ghcup
-RUN curl --proto '=https' --tlsv1.2 -sSf https://get-ghcup.haskell.org | BOOTSTRAP_HASKELL_NONINTERACTIVE=1 BOOTSTRAP_HASKELL_GHC_VERSION=8.10.7 BOOTSTRAP_HASKELL_CABAL_VERSION=3.6.2.0 sh
-
-# Adjust PATH
-ENV PATH="/root/.cabal/bin:/root/.ghcup/bin:$PATH"
-
-# Set both as default
-RUN ghcup set ghc 8.10.7 && \
-    ghcup set cabal
-
-COPY . /project
-WORKDIR /project
-
-# Compile smp-server
-RUN cabal update
-RUN cabal install
-
-### Final stage
-
-FROM final
-
-# Install OpenSSL dependency
-RUN apt-get update && apt-get install -y openssl libnuma-dev
-
-# Copy compiled smp-server from build stage
-COPY --from=build /root/.cabal/bin/smp-server /usr/bin/smp-server
-
-# Copy our helper script
-COPY ./scripts/docker/entrypoint /usr/bin/entrypoint
-
-# Open smp-server listening port
-EXPOSE 5223
-
-# SimpleX requires using SIGINT to correctly preserve undelivered messages and restore them on restart
-STOPSIGNAL SIGINT
-
-# Finally, execute helper script
-ENTRYPOINT [ "/usr/bin/entrypoint" ]
@@ -3,6 +3,14 @@ packages: .
 -- packages: . ../hs-socks
 -- packages: . ../http2

+package *
+  coverage: True
+  library-coverage: True
+
+package attoparsec
+  coverage: False
+  library-coverage: False
+
 source-repository-package
    type: git
    location: https://github.com/simplex-chat/aeson.git
@@ -1,20 +0,0 @@
-FROM ubuntu:focal
-
-# Install curl
-RUN apt-get update && apt-get install -y curl
-
-# Download latest smp-server release and assign executable permission
-RUN curl -L https://github.com/simplex-chat/simplexmq/releases/latest/download/smp-server-ubuntu-20_04-x86-64 -o /usr/bin/smp-server && \
-	chmod +x /usr/bin/smp-server
-
-# Copy our helper script
-COPY ./scripts/docker/entrypoint /usr/bin/entrypoint
-
-# Open smp-server listening port
-EXPOSE 5223
-
-# SimpleX requires using SIGINT to correctly preserve undelivered messages and restore them on restart
-STOPSIGNAL SIGINT
-
-# Finally, execute helper script
-ENTRYPOINT [ "/usr/bin/entrypoint" ]
@@ -0,0 +1,141 @@
+#!/usr/bin/env sh
+set -eu
+
+# Links to scripts/configs
+bin="https://github.com/simplex-chat/simplexmq/releases/latest/download"
+bin_smp="$bin/smp-server-ubuntu-20_04-x86-64"
+bin_xftp="$bin/xftp-server-ubuntu-20_04-x86-64"
+
+scripts="https://raw.githubusercontent.com/simplex-chat/simplexmq/stable/scripts/main"
+scripts_systemd_smp="$scripts/smp-server.service"
+scripts_systemd_xftp="$scripts/xftp-server.service"
+scripts_update="$scripts/simplex-servers-update"
+scripts_uninstall="$scripts/simplex-servers-uninstall"
+
+# Default installation paths
+path_bin="/usr/local/bin"
+path_bin_smp="$path_bin/smp-server"
+path_bin_xftp="$path_bin/xftp-server"
+path_bin_update="$path_bin/simplex-servers-update"
+path_bin_uninstall="$path_bin/simplex-servers-uninstall"
+
+path_conf_etc="/etc/opt"
+path_conf_var="/var/opt"
+path_conf_smp="$path_conf_etc/simplex $path_conf_var/simplex"
+path_conf_xftp="$path_conf_etc/simplex-xftp $path_conf_var/simplex-xftp /srv/xftp"
+
+path_systemd="/etc/systemd/system"
+path_systemd_smp="$path_systemd/smp-server.service"
+path_systemd_xftp="$path_systemd/xftp-server.service"
+
+# Defaut users
+user_smp="smp"
+user_xftp="xftp"
+
+GRN='\033[0;32m'
+BLU='\033[1;34m'
+YLW='\033[1;33m'
+RED='\033[0;31m'
+NC='\033[0m'
+
+logo='
+ ____  _                 _     __  __
+/ ___|(_)_ __ ___  _ __ | | ___\ \/ /
+\___ \| | '"'"'_ ` _ \| '"'"'_ \| |/ _ \\  / 
+ ___) | | | | | | | |_) | |  __//  \ 
+|____/|_|_| |_| |_| .__/|_|\___/_/\_\
+                  |_|                            
+'
+
+welcome="Welcome to SMP/XFTP installation script! Here's what we're going to do:
+${GRN}1.${NC} Install latest binaries from GitHub releases:
+    - smp: ${YLW}${path_bin_smp}${NC}
+    - xftp: ${YLW}${path_bin_xftp}${NC}
+${GRN}2.${NC} Create server directories:
+    - smp: ${YLW}${path_conf_smp}${NC}
+    - xftp: ${YLW}${path_conf_xftp}${NC}
+${GRN}3.${NC} Setup user for each server:
+    - xmp: ${YLW}${user_smp}${NC}
+    - xftp: ${YLW}${user_xftp}${NC}
+${GRN}4.${NC} Create systemd services:
+    - smp: ${YLW}${path_systemd_smp}${NC}
+    - xftp: ${YLW}${path_systemd_xftp}${NC}
+${GRN}5.${NC} Install update and uninstallation script:
+    - all: ${YLW}${path_bin_update}${NC}, ${YLW}${path_bin_uninstall}${NC}
+
+Press ${GRN}ENTER${NC} to continue or ${RED}Ctrl+C${NC} to cancel installation"
+
+end="Installtion is complete!
+
+Please checkout our server guides:
+- smp: ${GRN}https://simplex.chat/docs/server.html${NC}
+- xftp: ${GRN}https://simplex.chat/docs/xftp-server.html${NC}
+
+To uninstall with full clean-up, simply run: ${YLW}sudo /usr/local/bin/simplex-servers-uninstall${NC}
+"
+
+setup_bins() {
+ curl --proto '=https' --tlsv1.2 -sSf -L "$bin_smp" -o "$path_bin_smp" && chmod +x "$path_bin_smp"
+ curl --proto '=https' --tlsv1.2 -sSf -L "$bin_xftp" -o "$path_bin_xftp" && chmod +x "$path_bin_xftp"
+}
+
+setup_users() {
+ useradd -M "$user_smp" 2> /dev/null || true
+ useradd -M "$user_xftp" 2> /dev/null || true
+}
+
+setup_dirs() {
+ # Unquoted varibles, so field splitting can occur
+ mkdir -p $path_conf_smp
+ chown "$user_smp":"$user_smp" "$path_conf_smp"
+ mkdir -p $path_conf_xftp
+ chown "$user_xftp":"$user_xftp" "$path_conf_xftp"
+}
+
+setup_systemd() {
+ curl --proto '=https' --tlsv1.2 -sSf -L "$scripts_systemd_smp" -o "$path_systemd_smp"
+ curl --proto '=https' --tlsv1.2 -sSf -L "$scripts_systemd_xftp" -o "$path_systemd_xftp"
+}
+
+setup_scripts() {
+ curl --proto '=https' --tlsv1.2 -sSf -L "$scripts_update" -o "$path_bin_update" && chmod +x "$path_bin_update"
+ curl --proto '=https' --tlsv1.2 -sSf -L "$scripts_uninstall" -o "$path_bin_uninstall" && chmod +x "$path_bin_uninstall"
+ }
+
+checks() {
+ if [ "$(id -u)" -ne 0 ]; then
+  printf "This script is intended to be run with root privileges. Please re-run script using sudo."
+  exit 1
+ fi
+}
+
+main() {
+ checks
+ 
+ printf "%b\n%b\n" "${BLU}$logo${NC}" "$welcome"
+ read ans
+
+ printf "Installing binaries..."
+ setup_bins
+ printf "${GRN} Done!${NC}\n"
+
+ printf "Creating users..."
+ setup_users
+ printf "${GRN} Done!${NC}\n"
+ 
+ printf "Creating directories..."
+ setup_dirs
+ printf "${GRN} Done!${NC}\n"
+
+ printf "Creating systemd services..."
+ setup_systemd
+ printf "${GRN} Done!${NC}\n"
+
+ printf "Installing update and uninstallation script..."
+ setup_scripts
+ printf "${GRN} Done!${NC}\n"
+
+ printf "%b" "$end"
+}
+
+main
@@ -1,5 +1,5 @@
 name: simplexmq
-version: 5.0.0
+version: 5.1.1
 synopsis: SimpleXMQ message broker
 description: |
  This package includes <./docs/Simplex-Messaging-Server.html server>,
@@ -144,6 +144,8 @@ tests:
      - silently == 1.2.*
      - main-tester == 0.2.*
      - timeit == 2.0.*
+    ghc-options:
+      - -fhpc

 ghc-options:
  # - -haddock
@@ -1,29 +0,0 @@
-#!/usr/bin/env sh
-confd="/etc/opt/simplex"
-logd="/var/opt/simplex/"
-
-# 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 ;;
-    *[a-zA-Z]*) set -- -n $addr ;;
-    *) set -- --ip $addr ;;
-  esac
-
-  case $pass in
-    '') set -- "$@" --no-password ;;
-    *) set -- "$@" --password $pass ;;
-  esac
-
-  smp-server init -y -l "$@"
-fi
-
-# backup store log
-[ -f "$logd/smp-server-store.log" ] && cp "$logd"/smp-server-store.log "$logd"/smp-server-store.log.bak
-# rotate server log
-[ -f "$logd/smp-server.log" ] && mv "$logd"/smp-server.log "$logd"/smp-server-"$(date +'%FT%T')".log
-
-# 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 > "$logd"/smp-server.log 2>&1
@@ -0,0 +1,48 @@
+#!/usr/bin/env sh
+confd='/etc/opt/simplex'
+logd='/var/opt/simplex/'
+
+# 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 ;;
+    *[a-zA-Z]*)
+      case "${ADDR}" in
+        *:*) set -- --ip "${ADDR}" ;;
+        *) set -- -n "${ADDR}" ;;
+      esac
+      ;;
+    *) set -- --ip "${ADDR}" ;;
+  esac
+
+  # Optionally, set password
+  case "${PASS}" in
+    '') set -- "$@" --no-password ;;
+    *) set -- "$@" --password "${PASS}" ;;
+  esac
+
+  # And init certificates and configs
+  smp-server init -y -l "$@"
+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
+
+# 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
+
@@ -0,0 +1,48 @@
+#!/usr/bin/env sh
+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 ;;
+    *[a-zA-Z]*)
+      case "${ADDR}" in
+        *:*) set -- --ip "${ADDR}" ;;
+        *) set -- -n "${ADDR}" ;;
+      esac
+      ;;
+    *) set -- --ip "${ADDR}" ;;
+  esac
+
+  # Set quota
+  case "${QUOTA}" in
+    '') printf 'Please specify $QUOTA environment variable.\n'; exit 1 ;;
+    *) set -- "$@" --quota "${QUOTA}" ;;
+  esac
+
+  # Init the certificates and configs
+  xftp-server init -l -p /srv/xftp "$@"
+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
+
+# 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
+
@@ -0,0 +1,18 @@
+#!/usr/bin/env sh
+set -eu
+
+GRN='\033[0;32m'
+RED='\033[0;31m'
+NC='\033[0m'
+
+if [ "$(id -u)" -ne 0 ]; then
+  printf "This script is intended to be run with root privileges. Please re-run script using sudo."
+  exit 1
+fi
+
+printf "${RED}This action will permanently remove all configs, directories, binaries from Installation Script. Please backup any relevant configs if they are needed.${NC}\n\nPress ${GRN}ENTER${NC} to continue or ${RED}Ctrl+C${NC} to cancel installation"
+read ans
+
+rm -rf /var/opt/simplex /etc/opt/simplex /var/opt/simplex-xftp /etc/opt/simplex-xftp /srv/xftp /etc/systemd/system/smp-server.service /etc/systemd/system/xftp-server.service /usr/local/bin/smp-server /usr/local/bin/xftp-server /usr/local/bin/simplex-servers-update /usr/local/bin/simplex-servers-uninstall && userdel smp && userdel xftp
+
+printf "Uninstallation is complete! Thanks for trying out SimpleX!\n"
@@ -0,0 +1,160 @@
+#!/usr/bin/env sh
+set -eu
+
+# Links to scripts/configs
+bin="https://github.com/simplex-chat/simplexmq/releases/latest/download"
+bin_smp="$bin/smp-server-ubuntu-20_04-x86-64"
+bin_xftp="$bin/xftp-server-ubuntu-20_04-x86-64"
+
+scripts="https://raw.githubusercontent.com/simplex-chat/simplexmq/stable/scripts/main"
+scripts_systemd_smp="$scripts/smp-server.service"
+scripts_systemd_xftp="$scripts/xftp-server.service"
+scripts_update="$scripts/simplex-servers-update"
+scripts_uninstall="$scripts/simplex-servers-uninstall"
+
+# Default installation paths
+path_bin="/usr/local/bin"
+path_bin_smp="$path_bin/smp-server"
+path_bin_xftp="$path_bin/xftp-server"
+path_bin_update="$path_bin/simplex-servers-update"
+path_bin_uninstall="$path_bin/simplex-servers-uninstall"
+
+path_systemd="/etc/systemd/system"
+path_systemd_smp="$path_systemd/smp-server.service"
+path_systemd_xftp="$path_systemd/xftp-server.service"
+
+# Temporary paths
+path_tmp_bin="$(mktemp -d)"
+path_tmp_bin_update="$path_tmp_bin/simplex-servers-update"
+path_tmp_bin_uninstall="$path_tmp_bin/simplex-servers-uninstall"
+path_tmp_systemd_smp="$path_tmp_bin/smp-server.service"
+path_tmp_systemd_xftp="$path_tmp_bin/xftp-server.service"
+
+GRN='\033[0;32m'
+BLU='\033[1;36m'
+YLW='\033[1;33m'
+RED='\033[0;31m'
+NC='\033[0m'
+
+# Currently, XFTP default to v0.1.0, so it doesn't make sense to check its version
+local_version="$($path_bin_smp -v | awk '{print $3}')"
+remote_version="$(curl --proto '=https' --tlsv1.2 -sSf -L https://api.github.com/repos/simplex-chat/simplexmq/releases/latest | grep -i "tag_name" | awk -F \" '{print $4}')"
+
+update_scripts() {
+  curl --proto '=https' --tlsv1.2 -sSf -L "$scripts_update" -o "$path_tmp_bin_update" && chmod +x "$path_tmp_bin_update"
+  curl --proto '=https' --tlsv1.2 -sSf -L "$scripts_uninstall" -o "$path_tmp_bin_uninstall" && chmod +x "$path_tmp_bin_uninstall"
+
+  if diff -q "$path_bin_uninstall" "$path_tmp_bin_uninstall" > /dev/null; then
+    printf -- "- ${YLW}Uninstall script is up-to-date${NC}.\n"
+    rm "$path_tmp_bin_uninstall"
+  else
+    printf -- "- Updating uninstall script..."
+    mv "$path_tmp_bin_uninstall" "$path_bin_uninstall"
+    printf "${GRN}Done!${NC}\n"
+  fi
+  if diff -q "$path_bin_update" "$path_tmp_bin_update" > /dev/null; then
+    printf -- "- ${YLW}Update script is up-to-date${NC}.\n"
+    rm "$path_tmp_bin_update"
+  else
+    printf -- "- Updating update script..."
+    mv "$path_tmp_bin_update" "$path_bin_update"
+    printf "${GRN}Done!${NC}\n"
+    printf "Re-executing Update script with latest updates..."
+    exec sh "$path_bin_update" "continue"
+  fi
+}
+
+update_systemd() {
+  curl --proto '=https' --tlsv1.2 -sSf -L "$scripts_systemd_smp" -o "$path_tmp_systemd_smp"
+  curl --proto '=https' --tlsv1.2 -sSf -L "$scripts_systemd_xftp" -o "$path_tmp_systemd_xftp"
+
+  if diff -q "$path_systemd_smp" "$path_tmp_systemd_smp" > /dev/null; then
+    printf -- "- ${YLW}smp-server service is up-to-date${NC}.\n"
+    rm "$path_tmp_systemd_smp"
+  else
+    printf -- "- Updating smp-server service..."
+    mv "$path_tmp_systemd_smp" "$path_systemd_smp"
+    systemctl daemon-reload
+    printf "${GRN}Done!${NC}\n"
+  fi
+  if diff -q "$path_systemd_xftp" "$path_tmp_systemd_xftp" > /dev/null; then
+    printf -- "- ${YLW}xftp-server service is up-to-date${NC}.\n"
+    rm "$path_tmp_systemd_xftp"
+  else
+    printf -- "- Updating xftp-server service..."
+    mv "$path_tmp_systemd_xftp" "$path_systemd_xftp"
+    systemctl daemon-reload
+    printf "${GRN}Done!${NC}\n"
+  fi
+}
+
+update_bins() {
+  if [ "$local_version" != "$remote_version" ]; then
+    if systemctl is-active --quiet smp-server; then
+      printf -- "- Stopping smp-server service..."
+      systemctl stop smp-server
+      printf "${GRN}Done!${NC}\n"
+
+      print -- "- Updating smp-server bin to %s..." "$remote_version"
+      curl --proto '=https' --tlsv1.2 -sSf -L "$bin_smp" -o "$bin_path_smp" && chmod +x "$bin_path_smp"
+      printf "${GRN}Done!${NC}\n"
+      
+      printf -- "- Starting smp-server service..."
+      systemctl stop smp-server
+      printf "${GRN}Done!${NC}\n"
+    else
+      print -- "- Updating smp-server bin..."
+      curl --proto '=https' --tlsv1.2 -sSf -L "$bin_smp" -o "$bin_path_smp" && chmod +x "$bin_path_smp"
+      printf "${GRN}Done!${NC}\n"
+    fi
+
+    if systemctl is-active --quiet xftp-server; then
+      printf -- "- Stopping xftp-server service..."
+      systemctl stop xftp-server
+      printf "${GRN}Done!${NC}\n"
+
+      print -- "- Updating xftp-server bin to %s..." "$remote_version"
+      curl --proto '=https' --tlsv1.2 -sSf -L "$bin_xftp" -o "$bin_path_xftp" && chmod +x "$bin_path_xftp"
+      printf "${GRN}Done!${NC}\n"
+      
+      printf -- "- Starting xftp-server service..."
+      systemctl stop xftp-server
+      printf "${GRN}Done!${NC}\n"
+    else
+      print -- "- Updating xftp-server bin..."
+      curl --proto '=https' --tlsv1.2 -sSf -L "$bin_smp" -o "$bin_path_xftp" && chmod +x "$bin_path_xftp"
+      printf "${GRN}Done!${NC}\n"
+    fi
+  else
+    printf -- "- ${YLW}smp-server and xftp-server binaries is up-to-date${NC}.\n"
+  fi
+}
+
+checks() {
+  if [ "$(id -u)" -ne 0 ]; then
+    printf "This script is intended to be run with root privileges. Please re-run script using sudo.\n"
+    exit 1
+  fi
+}
+
+main() {
+  checks
+
+  set +u
+  if [ "$1" != "continue" ]; then
+    set -u
+    printf "Updating scripts...\n"
+    update_scripts
+  else
+    set -u
+    printf "${GRN}Done!${NC}\n"
+  fi
+
+  printf "Updating systemd services...\n"
+  update_systemd
+  
+  printf "Updating simplex server binaries...\n"
+  update_bins
+}
+
+main "$@"
@@ -0,0 +1,15 @@
+[Unit]
+Description=SMP server
+
+[Service]
+User=smp
+Group=smp
+Type=simple
+ExecStart=/usr/local/bin/smp-server start +RTS -N -RTS 
+ExecStopPost=/usr/bin/env sh -c '[ -e "/var/opt/simplex/smp-server-store.log" ] && cp "/var/opt/simplex/smp-server-store.log" "/var/opt/simplex/smp-server-store.log.$(date +%FT%T)'
+LimitNOFILE=65535
+KillSignal=SIGINT
+TimeoutStopSec=infinity
+
+[Install]
+WantedBy=multi-user.target
@@ -0,0 +1,16 @@
+[Unit]
+Description=XFTP server
+
+[Service]
+User=xftp
+Group=xftp
+Type=simple
+ExecStart=/usr/local/bin/xftp-server start +RTS -N -RTS
+ExecStopPost=/usr/bin/env sh -c '[ -e "/var/opt/simplex-xftp/file-server-store.log" ] && cp "/var/opt/simplex-xftp/file-server-store.log" "/var/opt/simplex-xftp/file-server-store.log.$(date +%FT%T)'
+LimitNOFILE=65535
+KillSignal=SIGINT
+TimeoutStopSec=infinity
+AmbientCapabilities=CAP_NET_BIND_SERVICE
+   
+[Install]
+WantedBy=multi-user.target
@@ -5,7 +5,7 @@ cabal-version: 1.12
 -- see: https://github.com/sol/hpack

 name:           simplexmq
-version:        5.0.0
+version:        5.1.1
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                <./docs/Simplex-Messaging-Client.html client> and
@@ -80,6 +80,8 @@ library
      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230223_files
      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230320_retry_state
      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230401_snd_files
+      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230510_files_pending_replicas_indexes
+      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230516_encrypted_rcv_message_hashes
      Simplex.Messaging.Agent.TAsyncs
      Simplex.Messaging.Agent.TRcvQueues
      Simplex.Messaging.Client
@@ -543,7 +545,7 @@ test-suite simplexmq-test
      Paths_simplexmq
  hs-source-dirs:
      tests
-  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns
+  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -fhpc
  build-depends:
      HUnit ==1.6.*
    , QuickCheck ==2.14.*
@@ -66,7 +66,7 @@ import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol (EntityId, XFTPServer, XFTPServerWithAuth)
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
-import Simplex.Messaging.Util (liftError, liftIOEither, tshow, whenM)
+import Simplex.Messaging.Util (liftError, liftIOEither, tshow, unlessM, whenM)
 import System.FilePath (takeFileName, (</>))
 import UnliftIO
 import UnliftIO.Concurrent
@@ -273,9 +273,9 @@ runXFTPRcvLocalWorker c doWork = do
        getChunkPaths (RcvFileChunk {chunkTmpPath = Nothing} : _cs) =
          throwError $ INTERNAL "no chunk path"

-deleteRcvFile :: AgentMonad m => AgentClient -> UserId -> RcvFileId -> m ()
-deleteRcvFile c userId rcvFileEntityId = do
-  RcvFile {rcvFileId, prefixPath, status} <- withStore c $ \db -> getRcvFileByEntityId db userId rcvFileEntityId
+deleteRcvFile :: AgentMonad m => AgentClient -> RcvFileId -> m ()
+deleteRcvFile c rcvFileEntityId = do
+  RcvFile {rcvFileId, prefixPath, status} <- withStore c $ \db -> getRcvFileByEntityId db rcvFileEntityId
  if status == RFSComplete || status == RFSError
    then do
      removePath prefixPath
@@ -415,7 +415,7 @@ runXFTPSndPrepareWorker c doWork = do
        createChunk :: Int -> SndFileChunk -> m ()
        createChunk numRecipients' ch = do
          atomically $ assertAgentForeground c
-          (replica, ProtoServerWithAuth srv _) <- agentOperationBracket c AOSndNetwork throwWhenInactive tryCreate
+          (replica, ProtoServerWithAuth srv _) <- tryCreate
          withStore' c $ \db -> createSndFileReplica db ch replica
          addXFTPSndWorker c $ Just srv
          where
@@ -445,7 +445,7 @@ runXFTPSndWorker c srv doWork = do
  forever $ do
    void . atomically $ readTMVar doWork
    atomically $ assertAgentForeground c
-    agentOperationBracket c AOSndNetwork throwWhenInactive runXFTPOperation
+    runXFTPOperation
  where
    noWorkToDo = void . atomically $ tryTakeTMVar doWork
    runXFTPOperation :: m ()
@@ -475,6 +475,7 @@ runXFTPSndWorker c srv doWork = do
    uploadFileChunk sndFileChunk@SndFileChunk {sndFileId, userId, chunkSpec = chunkSpec@XFTPChunkSpec {filePath}, digest = chunkDigest} replica = do
      replica'@SndFileChunkReplica {sndChunkReplicaId} <- addRecipients sndFileChunk replica
      fsFilePath <- toFSFilePath filePath
+      unlessM (doesFileExist fsFilePath) $ throwError $ INTERNAL "encrypted file doesn't exist on upload"
      let chunkSpec' = chunkSpec {filePath = fsFilePath} :: XFTPChunkSpec
      atomically $ assertAgentForeground c
      agentXFTPUploadChunk c userId chunkDigest replica' chunkSpec'
@@ -567,9 +568,9 @@ runXFTPSndWorker c srv doWork = do
        chunkUploaded SndFileChunk {replicas} =
          any (\SndFileChunkReplica {replicaStatus} -> replicaStatus == SFRSUploaded) replicas

-deleteSndFileInternal :: AgentMonad m => AgentClient -> UserId -> SndFileId -> m ()
-deleteSndFileInternal c userId sndFileEntityId = do
-  SndFile {sndFileId, prefixPath, status} <- withStore c $ \db -> getSndFileByEntityId db userId sndFileEntityId
+deleteSndFileInternal :: AgentMonad m => AgentClient -> SndFileId -> m ()
+deleteSndFileInternal c sndFileEntityId = do
+  SndFile {sndFileId, prefixPath, status} <- withStore c $ \db -> getSndFileByEntityId db sndFileEntityId
  if status == SFSComplete || status == SFSError
    then do
      forM_ prefixPath $ removePath <=< toFSFilePath
@@ -578,7 +579,7 @@ deleteSndFileInternal c userId sndFileEntityId = do

 deleteSndFileRemote :: forall m. AgentMonad m => AgentClient -> UserId -> SndFileId -> ValidFileDescription 'FSender -> m ()
 deleteSndFileRemote c userId sndFileEntityId (ValidFileDescription FileDescription {chunks}) = do
-  deleteSndFileInternal c userId sndFileEntityId `catchError` (notify c sndFileEntityId . SFERR)
+  deleteSndFileInternal c sndFileEntityId `catchError` (notify c sndFileEntityId . SFERR)
  forM_ chunks $ \ch -> deleteFileChunk ch `catchError` (notify c sndFileEntityId . SFERR)
  where
    deleteFileChunk :: FileChunk -> m ()
@@ -68,10 +68,13 @@ data XFTPEnv = XFTPEnv
    serverStats :: FileServerStats
  }

+defFileExpirationHours :: Int64
+defFileExpirationHours = 48
+
 defaultFileExpiration :: ExpirationConfig
 defaultFileExpiration =
  ExpirationConfig
-    { ttl = 48 * 3600, -- seconds, 48 hours
+    { ttl = defFileExpirationHours * 3600, -- seconds
      checkInterval = 2 * 3600 -- seconds, 2 hours
    }

@@ -18,11 +18,12 @@ import Network.Socket (HostName)
 import Options.Applicative
 import Simplex.FileTransfer.Description (FileSize (..), kb, mb)
 import Simplex.FileTransfer.Server (runXFTPServer)
-import Simplex.FileTransfer.Server.Env (XFTPServerConfig (..), defaultFileExpiration)
+import Simplex.FileTransfer.Server.Env (XFTPServerConfig (..), defaultFileExpiration, defFileExpirationHours)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol (ProtoServerWithAuth (..), pattern XFTPServer)
 import Simplex.Messaging.Server.CLI
+import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Transport.Client (TransportHost (..))
 import System.Directory (createDirectoryIfMissing, doesFileExist)
 import System.FilePath (combine)
@@ -30,7 +31,7 @@ import System.IO (BufferMode (..), hSetBuffering, stderr, stdout)
 import Text.Read (readMaybe)

 xftpServerVersion :: String
-xftpServerVersion = "0.1.0"
+xftpServerVersion = "1.0.0"

 xftpServerCLI :: FilePath -> FilePath -> IO ()
 xftpServerCLI cfgPath logPath = do
@@ -76,6 +77,8 @@ xftpServerCLI cfgPath logPath = do
          \# and restoring it when the server is started.\n\
          \# Log is compacted on start (deleted objects are removed).\n"
            <> ("enable: " <> onOff enableStoreLog <> "\n\n")
+            <> "# Expire files after the specified number of hours.\n"
+            <> ("expire_files_hours: " <> show defFileExpirationHours <> "\n\n")
            <> "log_stats: off\n\
               \\n\
               \[AUTH]\n\
@@ -113,10 +116,13 @@ xftpServerCLI cfgPath logPath = do
        enableStoreLog = settingIsOn "STORE_LOG" "enable" ini
        logStats = settingIsOn "STORE_LOG" "log_stats" ini
        c = combine cfgPath . ($ defaultX509Config)
-        printXFTPConfig XFTPServerConfig {allowNewFiles, newFileBasicAuth, xftpPort, storeLogFile} = do
+        printXFTPConfig XFTPServerConfig {allowNewFiles, newFileBasicAuth, xftpPort, storeLogFile, fileExpiration} = do
          putStrLn $ case storeLogFile of
            Just f -> "Store log: " <> f
            _ -> "Store log disabled."
+          putStrLn $ case fileExpiration of
+            Just ExpirationConfig {ttl} -> "expiring files after " <> showTTL ttl
+            _ -> "not expiring files"
          putStrLn $
            "Uploading new files "
              <> if allowNewFiles
@@ -134,7 +140,10 @@ xftpServerCLI cfgPath logPath = do
              allowedChunkSizes = [kb 256, mb 1, mb 4],
              allowNewFiles = fromMaybe True $ iniOnOff "AUTH" "new_files" ini,
              newFileBasicAuth = either error id <$> strDecodeIni "AUTH" "create_password" ini,
-              fileExpiration = Just defaultFileExpiration,
+              fileExpiration =
+                Just defaultFileExpiration
+                  { ttl = 3600 * readIniDefault defFileExpirationHours "STORE_LOG" "expire_files_hours" ini
+                  },
              caCertificateFile = c caCrtFile,
              privateKeyFile = c serverKeyFile,
              certificateFile = c serverCrtFile,
@@ -345,16 +345,16 @@ xftpReceiveFile :: AgentErrorMonad m => AgentClient -> UserId -> ValidFileDescri
 xftpReceiveFile c = withAgentEnv c .: receiveFile c

 -- | Delete XFTP rcv file (deletes work files from file system and db records)
-xftpDeleteRcvFile :: AgentErrorMonad m => AgentClient -> UserId -> RcvFileId -> m ()
-xftpDeleteRcvFile c = withAgentEnv c .: deleteRcvFile c
+xftpDeleteRcvFile :: AgentErrorMonad m => AgentClient -> RcvFileId -> m ()
+xftpDeleteRcvFile c = withAgentEnv c . deleteRcvFile c

 -- | Send XFTP file
 xftpSendFile :: AgentErrorMonad m => AgentClient -> UserId -> FilePath -> Int -> m SndFileId
 xftpSendFile c = withAgentEnv c .:. sendFile c

 -- | Delete XFTP snd file internally (deletes work files from file system and db records)
-xftpDeleteSndFileInternal :: AgentErrorMonad m => AgentClient -> UserId -> SndFileId -> m ()
-xftpDeleteSndFileInternal c = withAgentEnv c .: deleteSndFileInternal c
+xftpDeleteSndFileInternal :: AgentErrorMonad m => AgentClient -> SndFileId -> m ()
+xftpDeleteSndFileInternal c = withAgentEnv c . deleteSndFileInternal c

 -- | Delete XFTP snd file chunks on servers
 xftpDeleteSndFileRemote :: AgentErrorMonad m => AgentClient -> UserId -> SndFileId -> ValidFileDescription 'FSender -> m ()
@@ -1601,6 +1601,7 @@ cleanupManager c@AgentClient {subQ} = do
  forever $ do
    void . runExceptT $ do
      deleteConns `catchError` (notify "" . ERR)
+      deleteRcvMsgHashes `catchError` (notify "" . ERR)
      deleteRcvFilesExpired `catchError` (notify "" . RFERR)
      deleteRcvFilesDeleted `catchError` (notify "" . RFERR)
      deleteRcvFilesTmpPaths `catchError` (notify "" . RFERR)
@@ -1614,6 +1615,9 @@ cleanupManager c@AgentClient {subQ} = do
      withLock (deleteLock c) "cleanupManager" $ do
        void $ withStore' c getDeletedConnIds >>= deleteDeletedConns c
        withStore' c deleteUsersWithoutConns >>= mapM_ (notify "" . DEL_USER)
+    deleteRcvMsgHashes = do
+      rcvMsgHashesTTL <- asks $ rcvMsgHashesTTL . config
+      withStore' c (`deleteRcvMsgHashesExpired` rcvMsgHashesTTL)
    deleteRcvFilesExpired = do
      rcvFilesTTL <- asks $ rcvFilesTTL . config
      rcvExpired <- withStore' c (`getRcvFilesExpired` rcvFilesTTL)
@@ -1652,6 +1656,8 @@ cleanupManager c@AgentClient {subQ} = do
    notify :: forall e. AEntityI e => EntityId -> ACommand 'Agent e -> ExceptT AgentErrorType m ()
    notify entId cmd = atomically $ writeTBQueue subQ ("", entId, APC (sAEntity @e) cmd)

+-- | make sure to ACK or throw in each message processing branch
+-- it cannot be finally, unfortunately, as sometimes it needs to be ACK+DEL
 processSMPTransmission :: forall m. AgentMonad m => AgentClient -> ServerTransmission BrokerMsg -> m ()
 processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, sessId, rId, cmd) = do
  (rq, SomeConn _ conn) <- withStore c (\db -> getRcvConn db srv rId)
@@ -1697,7 +1703,8 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
                              enqueueCommand c "" connId (Just server) $ AInternalCommand $ ICQDelete rcvId
                            _ -> notify . ERR . AGENT $ A_QUEUE "replaced RcvQueue not found in connection"
                        _ -> pure ()
-                      tryError agentClientMsg >>= \case
+                      let encryptedMsgHash = C.sha256Hash encAgentMsg
+                      tryError (agentClientMsg encryptedMsgHash) >>= \case
                        Right (Just (msgId, msgMeta, aMessage)) -> case aMessage of
                          HELLO -> helloMsg >> ackDel msgId
                          REPLY cReq -> replyMsg cReq >> ackDel msgId
@@ -1728,11 +1735,15 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
                                    logServer "<--" c srv rId "MSG <MSG>"
                                    notify $ MSG msgMeta msgFlags body
                                  _ -> pure ()
-                            _ -> throwError e
-                        Left e -> throwError e
+                            _ -> checkDuplicateHash e encryptedMsgHash >> ack
+                        Left e -> checkDuplicateHash e encryptedMsgHash >> ack
                      where
-                        agentClientMsg :: m (Maybe (InternalId, MsgMeta, AMessage))
-                        agentClientMsg = withStore c $ \db -> runExceptT $ do
+                        checkDuplicateHash :: AgentErrorType -> ByteString -> m ()
+                        checkDuplicateHash e encryptedMsgHash =
+                          unlessM (withStore' c $ \db -> checkRcvMsgHashExists db connId encryptedMsgHash) $
+                            throwError e
+                        agentClientMsg :: ByteString -> m (Maybe (InternalId, MsgMeta, AMessage))
+                        agentClientMsg encryptedMsgHash = withStore c $ \db -> runExceptT $ do
                          agentMsgBody <- agentRatchetDecrypt db connId encAgentMsg
                          liftEither (parse smpP (SEAgentError $ AGENT A_MESSAGE) agentMsgBody) >>= \case
                            agentMsg@(AgentMessage APrivHeader {sndMsgId, prevMsgHash} aMessage) -> do
@@ -1744,7 +1755,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
                                  recipient = (unId internalId, internalTs)
                                  broker = (srvMsgId, systemToUTCTime srvTs)
                                  msgMeta = MsgMeta {integrity, recipient, broker, sndMsgId}
-                                  rcvMsg = RcvMsgData {msgMeta, msgType, msgFlags, msgBody = agentMsgBody, internalRcvId, internalHash, externalPrevSndHash = prevMsgHash}
+                                  rcvMsg = RcvMsgData {msgMeta, msgType, msgFlags, msgBody = agentMsgBody, internalRcvId, internalHash, externalPrevSndHash = prevMsgHash, encryptedMsgHash}
                              liftIO $ createRcvMsg db connId rq rcvMsg
                              pure $ Just (internalId, msgMeta, aMessage)
                            _ -> pure Nothing
@@ -115,7 +115,6 @@ import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Either (lefts, partitionEithers)
 import Data.Functor (($>))
-import Data.Int (Int64)
 import Data.List (deleteFirstsBy, foldl', partition, (\\))
 import Data.List.NonEmpty (NonEmpty (..), (<|))
 import qualified Data.List.NonEmpty as L
@@ -82,6 +82,7 @@ data AgentConfig = AgentConfig
    helloTimeout :: NominalDiffTime,
    initialCleanupDelay :: Int64,
    cleanupInterval :: Int64,
+    rcvMsgHashesTTL :: NominalDiffTime,
    rcvFilesTTL :: NominalDiffTime,
    sndFilesTTL :: NominalDiffTime,
    xftpNotifyErrsOnRetry :: Bool,
@@ -145,6 +146,7 @@ defaultAgentConfig =
      helloTimeout = 2 * nominalDay,
      initialCleanupDelay = 30 * 1000000, -- 30 seconds
      cleanupInterval = 30 * 60 * 1000000, -- 30 minutes
+      rcvMsgHashesTTL = 30 * nominalDay,
      rcvFilesTTL = 2 * nominalDay,
      sndFilesTTL = nominalDay,
      xftpNotifyErrsOnRetry = True,
@@ -26,6 +26,7 @@ import Data.Bifunctor (first)
 import qualified Data.Map.Strict as M
 import Data.Text (Text)
 import Data.Time (UTCTime, addUTCTime, getCurrentTime)
+import Data.Time.Clock (diffUTCTime)
 import Simplex.Messaging.Agent.Client
 import Simplex.Messaging.Agent.Env.SQLite
 import Simplex.Messaging.Agent.Protocol (ACommand (..), APartyCmd (..), AgentErrorType (..), BrokerErrorType (..), ConnId, NotificationsMode (..), SAEntity (..))
@@ -39,7 +40,7 @@ import Simplex.Messaging.Notifications.Types
 import Simplex.Messaging.Protocol (NtfServer, ProtocolServer, SMPServer, sameSrvAddr)
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
-import Simplex.Messaging.Util (diffInMicros, threadDelay', tshow, unlessM)
+import Simplex.Messaging.Util (diffToMicroseconds, threadDelay', tshow, unlessM)
 import System.Random (randomR)
 import UnliftIO
 import UnliftIO.Concurrent (forkIO, threadDelay)
@@ -291,7 +292,7 @@ rescheduleAction doWork ts actionTs
  | otherwise = do
    void . atomically $ tryTakeTMVar doWork
    void . forkIO $ do
-      liftIO $ threadDelay' $ diffInMicros actionTs ts
+      liftIO $ threadDelay' $ diffToMicroseconds $ diffUTCTime actionTs ts
      void . atomically $ tryPutTMVar doWork ()
    pure True

@@ -428,7 +428,8 @@ data RcvMsgData = RcvMsgData
    msgBody :: MsgBody,
    internalRcvId :: InternalRcvId,
    internalHash :: MsgHash,
-    externalPrevSndHash :: MsgHash
+    externalPrevSndHash :: MsgHash,
+    encryptedMsgHash :: MsgHash
  }

 data RcvMsg = RcvMsg
@@ -94,8 +94,10 @@ module Simplex.Messaging.Agent.Store.SQLite
    deletePendingMsgs,
    setMsgUserAck,
    getLastMsg,
+    checkRcvMsgHashExists,
    deleteMsg,
    deleteSndMsgDelivery,
+    deleteRcvMsgHashesExpired,
    -- Double ratchet persistence
    createRatchetX3dhKeys,
    getRatchetX3dhKeys,
@@ -182,6 +184,7 @@ module Simplex.Messaging.Agent.Store.SQLite
    -- * utilities
    withConnection,
    withTransaction,
+    withTransactionCtx,
    firstRow,
    firstRow',
    maybeFirstRow,
@@ -212,7 +215,7 @@ import Data.Ord (Down (..))
 import Data.Text (Text)
 import qualified Data.Text as T
 import Data.Text.Encoding (decodeLatin1, encodeUtf8)
-import Data.Time.Clock (NominalDiffTime, UTCTime, addUTCTime, getCurrentTime)
+import Data.Time.Clock (NominalDiffTime, UTCTime, addUTCTime, diffUTCTime, getCurrentTime)
 import Data.Word (Word32)
 import Database.SQLite.Simple (FromRow, NamedParam (..), Only (..), Query (..), SQLError, ToRow, field, (:.) (..))
 import qualified Database.SQLite.Simple as DB
@@ -241,7 +244,7 @@ import Simplex.Messaging.Parsers (blobFieldParser, dropPrefix, fromTextField_, s
 import Simplex.Messaging.Protocol
 import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.Transport.Client (TransportHost)
-import Simplex.Messaging.Util (bshow, eitherToMaybe, ($>>=), (<$$>))
+import Simplex.Messaging.Util (bshow, diffToMilliseconds, eitherToMaybe, ($>>=), (<$$>))
 import Simplex.Messaging.Version
 import System.Directory (copyFile, createDirectoryIfMissing, doesFileExist)
 import System.Exit (exitFailure)
@@ -428,16 +431,29 @@ withConnection SQLiteStore {dbConnection} =
    (atomically . putTMVar dbConnection)

 withTransaction :: forall a. SQLiteStore -> (DB.Connection -> IO a) -> IO a
-withTransaction st action = withConnection st $ loop 500 3_000_000
+withTransaction = withTransactionCtx Nothing
+
+withTransactionCtx :: forall a. Maybe String -> SQLiteStore -> (DB.Connection -> IO a) -> IO a
+withTransactionCtx ctx_ st action = withConnection st $ loop 500 3_000_000
  where
    loop :: Int -> Int -> DB.Connection -> IO a
    loop t tLim db =
-      DB.withImmediateTransaction db (action db) `E.catch` \(e :: SQLError) ->
+      transactionWithCtx `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
+      where
+        transactionWithCtx = case ctx_ of
+          Nothing -> DB.withImmediateTransaction db (action db)
+          Just ctx -> do
+            t1 <- getCurrentTime
+            r <- DB.withImmediateTransaction db (action db)
+            t2 <- getCurrentTime
+            putStrLn $ "withTransactionCtx start :: " <> show t1 <> " :: " <> ctx
+            putStrLn $ "withTransactionCtx end   :: " <> show t2 <> " :: " <> ctx <> " :: duration=" <> show (diffToMilliseconds $ diffUTCTime t2 t1)
+            pure r

 createUserRecord :: DB.Connection -> IO UserId
 createUserRecord db = do
@@ -925,6 +941,17 @@ getLastMsg db connId msgId =
      let msgMeta = MsgMeta {recipient = (agentMsgId, internalTs), broker = (brokerId, brokerTs), sndMsgId, integrity}
       in RcvMsg {internalId = InternalId agentMsgId, msgMeta, msgBody, userAck}

+checkRcvMsgHashExists :: DB.Connection -> ConnId -> ByteString -> IO Bool
+checkRcvMsgHashExists db connId hash = do
+  fromMaybe False
+    <$> maybeFirstRow
+      fromOnly
+      ( DB.query
+          db
+          "SELECT 1 FROM encrypted_rcv_message_hashes WHERE conn_id = ? AND hash = ? LIMIT 1"
+          (connId, hash)
+      )
+
 deleteMsg :: DB.Connection -> ConnId -> InternalId -> IO ()
 deleteMsg db connId msgId =
  DB.execute db "DELETE FROM messages WHERE conn_id = ? AND internal_id = ?;" (connId, msgId)
@@ -938,6 +965,11 @@ deleteSndMsgDelivery db connId SndQueue {dbQueueId} msgId = do
  (Only (cnt :: Int) : _) <- DB.query db "SELECT count(*) FROM snd_message_deliveries WHERE conn_id = ? AND internal_id = ?" (connId, msgId)
  when (cnt == 0) $ deleteMsg db connId msgId

+deleteRcvMsgHashesExpired :: DB.Connection -> NominalDiffTime -> IO ()
+deleteRcvMsgHashesExpired db ttl = do
+  cutoffTs <- addUTCTime (- ttl) <$> getCurrentTime
+  DB.execute db "DELETE FROM encrypted_rcv_message_hashes WHERE created_at < ?" (Only cutoffTs)
+
 createRatchetX3dhKeys :: DB.Connection -> ConnId -> C.PrivateKeyX448 -> C.PrivateKeyX448 -> IO ()
 createRatchetX3dhKeys db connId x3dhPrivKey1 x3dhPrivKey2 =
  DB.execute db "INSERT INTO ratchets (conn_id, x3dh_priv_key_1, x3dh_priv_key_2) VALUES (?, ?, ?)" (connId, x3dhPrivKey1, x3dhPrivKey2)
@@ -1690,10 +1722,10 @@ insertRcvMsgBase_ dbConn connId RcvMsgData {msgMeta, msgType, msgFlags, msgBody,
    ]

 insertRcvMsgDetails_ :: DB.Connection -> ConnId -> RcvQueue -> RcvMsgData -> IO ()
-insertRcvMsgDetails_ dbConn connId RcvQueue {dbQueueId} RcvMsgData {msgMeta, internalRcvId, internalHash, externalPrevSndHash} = do
+insertRcvMsgDetails_ db connId RcvQueue {dbQueueId} RcvMsgData {msgMeta, internalRcvId, internalHash, externalPrevSndHash, encryptedMsgHash} = do
  let MsgMeta {integrity, recipient, broker, sndMsgId} = msgMeta
  DB.executeNamed
-    dbConn
+    db
    [sql|
      INSERT INTO rcv_messages
        ( conn_id, rcv_queue_id, internal_rcv_id, internal_id, external_snd_id,
@@ -1715,6 +1747,7 @@ insertRcvMsgDetails_ dbConn connId RcvQueue {dbQueueId} RcvMsgData {msgMeta, int
      ":external_prev_snd_hash" := externalPrevSndHash,
      ":integrity" := integrity
    ]
+  DB.execute db "INSERT INTO encrypted_rcv_message_hashes (conn_id, hash) VALUES (?,?)" (connId, encryptedMsgHash)

 updateHashRcv_ :: DB.Connection -> ConnId -> RcvMsgData -> IO ()
 updateHashRcv_ dbConn connId RcvMsgData {msgMeta, internalHash, internalRcvId} =
@@ -1888,15 +1921,15 @@ createRcvFile db gVar userId fd@FileDescription {chunks} prefixPath tmpPath save
        "INSERT INTO rcv_file_chunk_replicas (replica_number, rcv_file_chunk_id, xftp_server_id, replica_id, replica_key) VALUES (?,?,?,?,?)"
        (replicaNo, chunkId, srvId, replicaId, replicaKey)

-getRcvFileByEntityId :: DB.Connection -> UserId -> RcvFileId -> IO (Either StoreError RcvFile)
-getRcvFileByEntityId db userId rcvFileEntityId = runExceptT $ do
-  rcvFileId <- ExceptT $ getRcvFileIdByEntityId_ db userId rcvFileEntityId
+getRcvFileByEntityId :: DB.Connection -> RcvFileId -> IO (Either StoreError RcvFile)
+getRcvFileByEntityId db rcvFileEntityId = runExceptT $ do
+  rcvFileId <- ExceptT $ getRcvFileIdByEntityId_ db rcvFileEntityId
  ExceptT $ getRcvFile db rcvFileId

-getRcvFileIdByEntityId_ :: DB.Connection -> UserId -> RcvFileId -> IO (Either StoreError DBRcvFileId)
-getRcvFileIdByEntityId_ db userId rcvFileEntityId =
+getRcvFileIdByEntityId_ :: DB.Connection -> RcvFileId -> IO (Either StoreError DBRcvFileId)
+getRcvFileIdByEntityId_ db rcvFileEntityId =
  firstRow fromOnly SEFileNotFound $
-    DB.query db "SELECT rcv_file_id FROM rcv_files WHERE user_id = ? AND rcv_file_entity_id = ?" (userId, rcvFileEntityId)
+    DB.query db "SELECT rcv_file_id FROM rcv_files WHERE rcv_file_entity_id = ?" (Only rcvFileEntityId)

 getRcvFile :: DB.Connection -> DBRcvFileId -> IO (Either StoreError RcvFile)
 getRcvFile db rcvFileId = runExceptT $ do
@@ -2115,15 +2148,15 @@ createSndFile db gVar userId numRecipients path prefixPath key nonce =
      "INSERT INTO snd_files (snd_file_entity_id, user_id, num_recipients, key, nonce, path, prefix_path, status) VALUES (?,?,?,?,?,?,?,?)"
      (sndFileEntityId, userId, numRecipients, key, nonce, path, prefixPath, SFSNew)

-getSndFileByEntityId :: DB.Connection -> UserId -> SndFileId -> IO (Either StoreError SndFile)
-getSndFileByEntityId db userId sndFileEntityId = runExceptT $ do
-  sndFileId <- ExceptT $ getSndFileIdByEntityId_ db userId sndFileEntityId
+getSndFileByEntityId :: DB.Connection -> SndFileId -> IO (Either StoreError SndFile)
+getSndFileByEntityId db sndFileEntityId = runExceptT $ do
+  sndFileId <- ExceptT $ getSndFileIdByEntityId_ db sndFileEntityId
  ExceptT $ getSndFile db sndFileId

-getSndFileIdByEntityId_ :: DB.Connection -> UserId -> SndFileId -> IO (Either StoreError DBSndFileId)
-getSndFileIdByEntityId_ db userId sndFileEntityId =
+getSndFileIdByEntityId_ :: DB.Connection -> SndFileId -> IO (Either StoreError DBSndFileId)
+getSndFileIdByEntityId_ db sndFileEntityId =
  firstRow fromOnly SEFileNotFound $
-    DB.query db "SELECT snd_file_id FROM snd_files WHERE user_id = ? AND snd_file_entity_id = ?" (userId, sndFileEntityId)
+    DB.query db "SELECT snd_file_id FROM snd_files WHERE snd_file_entity_id = ?" (Only sndFileEntityId)

 getSndFile :: DB.Connection -> DBSndFileId -> IO (Either StoreError SndFile)
 getSndFile db sndFileId = runExceptT $ do
@@ -58,6 +58,8 @@ import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230217_server_key_hash
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230223_files
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230320_retry_state
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230401_snd_files
+import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230510_files_pending_replicas_indexes
+import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230516_encrypted_rcv_message_hashes
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (dropPrefix, sumTypeJSON)
 import Simplex.Messaging.Transport.Client (TransportHost)
@@ -82,7 +84,9 @@ schemaMigrations =
    ("m20230217_server_key_hash", m20230217_server_key_hash, Nothing),
    ("m20230223_files", m20230223_files, Just down_m20230223_files),
    ("m20230320_retry_state", m20230320_retry_state, Just down_m20230320_retry_state),
-    ("m20230401_snd_files", m20230401_snd_files, Just down_m20230401_snd_files)
+    ("m20230401_snd_files", m20230401_snd_files, Just down_m20230401_snd_files),
+    ("m20230510_files_pending_replicas_indexes", m20230510_files_pending_replicas_indexes, Just down_m20230510_files_pending_replicas_indexes),
+    ("m20230516_encrypted_rcv_message_hashes", m20230516_encrypted_rcv_message_hashes, Just down_m20230516_encrypted_rcv_message_hashes)
  ]

 -- | The list of migrations in ascending order by date
@@ -101,7 +105,8 @@ getCurrent db = map toMigration <$> DB.query_ db "SELECT name, down FROM migrati

 run :: Connection -> MigrationsToRun -> IO ()
 run db = \case
-  MTRUp ms -> mapM_ runUp ms
+  MTRUp [] -> pure ()
+  MTRUp ms -> mapM_ runUp ms >> execSQL "VACUUM;"
  MTRDown ms -> mapM_ runDown $ reverse ms
  MTRNone -> pure ()
  where
@@ -0,0 +1,22 @@
+{-# LANGUAGE QuasiQuotes #-}
+
+module Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230510_files_pending_replicas_indexes where
+
+import Database.SQLite.Simple (Query)
+import Database.SQLite.Simple.QQ (sql)
+
+m20230510_files_pending_replicas_indexes :: Query
+m20230510_files_pending_replicas_indexes =
+  [sql|
+CREATE INDEX idx_rcv_file_chunk_replicas_pending ON rcv_file_chunk_replicas(received, replica_number);
+CREATE INDEX idx_snd_file_chunk_replicas_pending ON snd_file_chunk_replicas(replica_status, replica_number);
+CREATE INDEX idx_deleted_snd_chunk_replicas_pending ON deleted_snd_chunk_replicas(created_at);
+|]
+
+down_m20230510_files_pending_replicas_indexes :: Query
+down_m20230510_files_pending_replicas_indexes =
+  [sql|
+DROP INDEX idx_deleted_snd_chunk_replicas_pending;
+DROP INDEX idx_snd_file_chunk_replicas_pending;
+DROP INDEX idx_rcv_file_chunk_replicas_pending;
+|]
@@ -0,0 +1,28 @@
+{-# LANGUAGE QuasiQuotes #-}
+
+module Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230516_encrypted_rcv_message_hashes where
+
+import Database.SQLite.Simple (Query)
+import Database.SQLite.Simple.QQ (sql)
+
+m20230516_encrypted_rcv_message_hashes :: Query
+m20230516_encrypted_rcv_message_hashes =
+  [sql|
+CREATE TABLE encrypted_rcv_message_hashes(
+  encrypted_rcv_message_hash_id INTEGER PRIMARY KEY,
+  conn_id BLOB NOT NULL REFERENCES connections ON DELETE CASCADE,
+  hash BLOB NOT NULL,
+  created_at TEXT NOT NULL DEFAULT (datetime('now')),
+  updated_at TEXT NOT NULL DEFAULT (datetime('now'))
+);
+
+CREATE INDEX idx_encrypted_rcv_message_hashes_hash ON encrypted_rcv_message_hashes(conn_id, hash);
+|]
+
+down_m20230516_encrypted_rcv_message_hashes :: Query
+down_m20230516_encrypted_rcv_message_hashes =
+  [sql|
+DROP INDEX idx_encrypted_rcv_message_hashes_hash;
+
+DROP TABLE encrypted_rcv_message_hashes;
+|]
@@ -188,7 +188,6 @@ tkn_dh_secret BLOB, -- DH secret for e2e encryption of notifications
  FOREIGN KEY(ntf_host, ntf_port) REFERENCES ntf_servers
  ON DELETE RESTRICT ON UPDATE CASCADE
 ) WITHOUT ROWID;
-CREATE UNIQUE INDEX idx_rcv_queues_ntf ON rcv_queues(host, port, ntf_id);
 CREATE TABLE ntf_subscriptions(
  conn_id BLOB NOT NULL,
  smp_host TEXT NULL,
@@ -224,8 +223,6 @@ CREATE TABLE commands(
  FOREIGN KEY(host, port) REFERENCES servers
  ON DELETE RESTRICT ON UPDATE CASCADE
 );
-CREATE UNIQUE INDEX idx_rcv_queue_id ON rcv_queues(conn_id, rcv_queue_id);
-CREATE UNIQUE INDEX idx_snd_queue_id ON snd_queues(conn_id, snd_queue_id);
 CREATE TABLE snd_message_deliveries(
  snd_message_delivery_id INTEGER PRIMARY KEY AUTOINCREMENT,
  conn_id BLOB NOT NULL REFERENCES connections ON DELETE CASCADE,
@@ -234,15 +231,135 @@ CREATE TABLE snd_message_deliveries(
  FOREIGN KEY(conn_id, internal_id) REFERENCES messages ON DELETE CASCADE DEFERRABLE INITIALLY DEFERRED
 );
 CREATE TABLE sqlite_sequence(name,seq);
-CREATE INDEX idx_snd_message_deliveries ON snd_message_deliveries(
-  conn_id,
-  snd_queue_id
-);
 CREATE TABLE users(
  user_id INTEGER PRIMARY KEY AUTOINCREMENT
  ,
  deleted INTEGER DEFAULT 0 CHECK(deleted NOT NULL)
 );
+CREATE TABLE xftp_servers(
+  xftp_server_id INTEGER PRIMARY KEY,
+  xftp_host TEXT NOT NULL,
+  xftp_port TEXT NOT NULL,
+  xftp_key_hash BLOB NOT NULL,
+  created_at TEXT NOT NULL DEFAULT(datetime('now')),
+  updated_at TEXT NOT NULL DEFAULT(datetime('now')),
+  UNIQUE(xftp_host, xftp_port, xftp_key_hash)
+);
+CREATE TABLE rcv_files(
+  rcv_file_id INTEGER PRIMARY KEY,
+  rcv_file_entity_id BLOB NOT NULL,
+  user_id INTEGER NOT NULL REFERENCES users ON DELETE CASCADE,
+  size INTEGER NOT NULL,
+  digest BLOB NOT NULL,
+  key BLOB NOT NULL,
+  nonce BLOB NOT NULL,
+  chunk_size INTEGER NOT NULL,
+  prefix_path TEXT NOT NULL,
+  tmp_path TEXT,
+  save_path TEXT NOT NULL,
+  status TEXT NOT NULL,
+  deleted INTEGER NOT NULL DEFAULT 0,
+  error TEXT,
+  created_at TEXT NOT NULL DEFAULT(datetime('now')),
+  updated_at TEXT NOT NULL DEFAULT(datetime('now')),
+  UNIQUE(rcv_file_entity_id)
+);
+CREATE TABLE rcv_file_chunks(
+  rcv_file_chunk_id INTEGER PRIMARY KEY,
+  rcv_file_id INTEGER NOT NULL REFERENCES rcv_files ON DELETE CASCADE,
+  chunk_no INTEGER NOT NULL,
+  chunk_size INTEGER NOT NULL,
+  digest BLOB NOT NULL,
+  tmp_path TEXT,
+  created_at TEXT NOT NULL DEFAULT(datetime('now')),
+  updated_at TEXT NOT NULL DEFAULT(datetime('now'))
+);
+CREATE TABLE rcv_file_chunk_replicas(
+  rcv_file_chunk_replica_id INTEGER PRIMARY KEY,
+  rcv_file_chunk_id INTEGER NOT NULL REFERENCES rcv_file_chunks ON DELETE CASCADE,
+  replica_number INTEGER NOT NULL,
+  xftp_server_id INTEGER NOT NULL REFERENCES xftp_servers ON DELETE CASCADE,
+  replica_id BLOB NOT NULL,
+  replica_key BLOB NOT NULL,
+  received INTEGER NOT NULL DEFAULT 0,
+  delay INTEGER,
+  retries INTEGER NOT NULL DEFAULT 0,
+  created_at TEXT NOT NULL DEFAULT(datetime('now')),
+  updated_at TEXT NOT NULL DEFAULT(datetime('now'))
+);
+CREATE TABLE snd_files(
+  snd_file_id INTEGER PRIMARY KEY,
+  snd_file_entity_id BLOB NOT NULL,
+  user_id INTEGER NOT NULL REFERENCES users ON DELETE CASCADE,
+  num_recipients INTEGER NOT NULL,
+  digest BLOB,
+  key BLOB NOT NUll,
+  nonce BLOB NOT NUll,
+  path TEXT NOT NULL,
+  prefix_path TEXT,
+  status TEXT NOT NULL,
+  deleted INTEGER NOT NULL DEFAULT 0,
+  error TEXT,
+  created_at TEXT NOT NULL DEFAULT(datetime('now')),
+  updated_at TEXT NOT NULL DEFAULT(datetime('now'))
+);
+CREATE TABLE snd_file_chunks(
+  snd_file_chunk_id INTEGER PRIMARY KEY,
+  snd_file_id INTEGER NOT NULL REFERENCES snd_files ON DELETE CASCADE,
+  chunk_no INTEGER NOT NULL,
+  chunk_offset INTEGER NOT NULL,
+  chunk_size INTEGER NOT NULL,
+  digest BLOB NOT NULL,
+  created_at TEXT NOT NULL DEFAULT(datetime('now')),
+  updated_at TEXT NOT NULL DEFAULT(datetime('now'))
+);
+CREATE TABLE snd_file_chunk_replicas(
+  snd_file_chunk_replica_id INTEGER PRIMARY KEY,
+  snd_file_chunk_id INTEGER NOT NULL REFERENCES snd_file_chunks ON DELETE CASCADE,
+  replica_number INTEGER NOT NULL,
+  xftp_server_id INTEGER NOT NULL REFERENCES xftp_servers ON DELETE CASCADE,
+  replica_id BLOB NOT NULL,
+  replica_key BLOB NOT NULL,
+  replica_status TEXT NOT NULL,
+  delay INTEGER,
+  retries INTEGER NOT NULL DEFAULT 0,
+  created_at TEXT NOT NULL DEFAULT(datetime('now')),
+  updated_at TEXT NOT NULL DEFAULT(datetime('now'))
+);
+CREATE TABLE snd_file_chunk_replica_recipients(
+  snd_file_chunk_replica_recipient_id INTEGER PRIMARY KEY,
+  snd_file_chunk_replica_id INTEGER NOT NULL REFERENCES snd_file_chunk_replicas ON DELETE CASCADE,
+  rcv_replica_id BLOB NOT NULL,
+  rcv_replica_key BLOB NOT NULL,
+  created_at TEXT NOT NULL DEFAULT(datetime('now')),
+  updated_at TEXT NOT NULL DEFAULT(datetime('now'))
+);
+CREATE TABLE deleted_snd_chunk_replicas(
+  deleted_snd_chunk_replica_id INTEGER PRIMARY KEY,
+  user_id INTEGER NOT NULL REFERENCES users ON DELETE CASCADE,
+  xftp_server_id INTEGER NOT NULL REFERENCES xftp_servers ON DELETE CASCADE,
+  replica_id BLOB NOT NULL,
+  replica_key BLOB NOT NULL,
+  chunk_digest BLOB NOT NULL,
+  delay INTEGER,
+  retries INTEGER NOT NULL DEFAULT 0,
+  created_at TEXT NOT NULL DEFAULT(datetime('now')),
+  updated_at TEXT NOT NULL DEFAULT(datetime('now'))
+);
+CREATE TABLE encrypted_rcv_message_hashes(
+  encrypted_rcv_message_hash_id INTEGER PRIMARY KEY,
+  conn_id BLOB NOT NULL REFERENCES connections ON DELETE CASCADE,
+  hash BLOB NOT NULL,
+  created_at TEXT NOT NULL DEFAULT(datetime('now')),
+  updated_at TEXT NOT NULL DEFAULT(datetime('now'))
+);
+CREATE UNIQUE INDEX idx_rcv_queues_ntf ON rcv_queues(host, port, ntf_id);
+CREATE UNIQUE INDEX idx_rcv_queue_id ON rcv_queues(conn_id, rcv_queue_id);
+CREATE UNIQUE INDEX idx_snd_queue_id ON snd_queues(conn_id, snd_queue_id);
+CREATE INDEX idx_snd_message_deliveries ON snd_message_deliveries(
+  conn_id,
+  snd_queue_id
+);
 CREATE INDEX idx_connections_user ON connections(user_id);
 CREATE INDEX idx_commands_conn_id ON commands(conn_id);
 CREATE INDEX idx_commands_host_port ON commands(host, port);
@@ -286,138 +403,43 @@ CREATE INDEX idx_snd_messages_conn_id_internal_id ON snd_messages(
  internal_id
 );
 CREATE INDEX idx_snd_queues_host_port ON snd_queues(host, port);
-CREATE TABLE xftp_servers(
-  xftp_server_id INTEGER PRIMARY KEY,
-  xftp_host TEXT NOT NULL,
-  xftp_port TEXT NOT NULL,
-  xftp_key_hash BLOB NOT NULL,
-  created_at TEXT NOT NULL DEFAULT(datetime('now')),
-  updated_at TEXT NOT NULL DEFAULT(datetime('now')),
-  UNIQUE(xftp_host, xftp_port, xftp_key_hash)
-);
-CREATE TABLE rcv_files(
-  rcv_file_id INTEGER PRIMARY KEY,
-  rcv_file_entity_id BLOB NOT NULL,
-  user_id INTEGER NOT NULL REFERENCES users ON DELETE CASCADE,
-  size INTEGER NOT NULL,
-  digest BLOB NOT NULL,
-  key BLOB NOT NULL,
-  nonce BLOB NOT NULL,
-  chunk_size INTEGER NOT NULL,
-  prefix_path TEXT NOT NULL,
-  tmp_path TEXT,
-  save_path TEXT NOT NULL,
-  status TEXT NOT NULL,
-  deleted INTEGER NOT NULL DEFAULT 0,
-  error TEXT,
-  created_at TEXT NOT NULL DEFAULT(datetime('now')),
-  updated_at TEXT NOT NULL DEFAULT(datetime('now')),
-  UNIQUE(rcv_file_entity_id)
-);
 CREATE INDEX idx_rcv_files_user_id ON rcv_files(user_id);
-CREATE TABLE rcv_file_chunks(
-  rcv_file_chunk_id INTEGER PRIMARY KEY,
-  rcv_file_id INTEGER NOT NULL REFERENCES rcv_files ON DELETE CASCADE,
-  chunk_no INTEGER NOT NULL,
-  chunk_size INTEGER NOT NULL,
-  digest BLOB NOT NULL,
-  tmp_path TEXT,
-  created_at TEXT NOT NULL DEFAULT(datetime('now')),
-  updated_at TEXT NOT NULL DEFAULT(datetime('now'))
-);
 CREATE INDEX idx_rcv_file_chunks_rcv_file_id ON rcv_file_chunks(rcv_file_id);
-CREATE TABLE rcv_file_chunk_replicas(
-  rcv_file_chunk_replica_id INTEGER PRIMARY KEY,
-  rcv_file_chunk_id INTEGER NOT NULL REFERENCES rcv_file_chunks ON DELETE CASCADE,
-  replica_number INTEGER NOT NULL,
-  xftp_server_id INTEGER NOT NULL REFERENCES xftp_servers ON DELETE CASCADE,
-  replica_id BLOB NOT NULL,
-  replica_key BLOB NOT NULL,
-  received INTEGER NOT NULL DEFAULT 0,
-  delay INTEGER,
-  retries INTEGER NOT NULL DEFAULT 0,
-  created_at TEXT NOT NULL DEFAULT(datetime('now')),
-  updated_at TEXT NOT NULL DEFAULT(datetime('now'))
-);
 CREATE INDEX idx_rcv_file_chunk_replicas_rcv_file_chunk_id ON rcv_file_chunk_replicas(
  rcv_file_chunk_id
 );
 CREATE INDEX idx_rcv_file_chunk_replicas_xftp_server_id ON rcv_file_chunk_replicas(
  xftp_server_id
 );
-CREATE TABLE snd_files(
-  snd_file_id INTEGER PRIMARY KEY,
-  snd_file_entity_id BLOB NOT NULL,
-  user_id INTEGER NOT NULL REFERENCES users ON DELETE CASCADE,
-  num_recipients INTEGER NOT NULL,
-  digest BLOB,
-  key BLOB NOT NUll,
-  nonce BLOB NOT NUll,
-  path TEXT NOT NULL,
-  prefix_path TEXT,
-  status TEXT NOT NULL,
-  deleted INTEGER NOT NULL DEFAULT 0,
-  error TEXT,
-  created_at TEXT NOT NULL DEFAULT(datetime('now')),
-  updated_at TEXT NOT NULL DEFAULT(datetime('now'))
-);
 CREATE INDEX idx_snd_files_user_id ON snd_files(user_id);
-CREATE TABLE snd_file_chunks(
-  snd_file_chunk_id INTEGER PRIMARY KEY,
-  snd_file_id INTEGER NOT NULL REFERENCES snd_files ON DELETE CASCADE,
-  chunk_no INTEGER NOT NULL,
-  chunk_offset INTEGER NOT NULL,
-  chunk_size INTEGER NOT NULL,
-  digest BLOB NOT NULL,
-  created_at TEXT NOT NULL DEFAULT(datetime('now')),
-  updated_at TEXT NOT NULL DEFAULT(datetime('now'))
-);
 CREATE INDEX idx_snd_file_chunks_snd_file_id ON snd_file_chunks(snd_file_id);
-CREATE TABLE snd_file_chunk_replicas(
-  snd_file_chunk_replica_id INTEGER PRIMARY KEY,
-  snd_file_chunk_id INTEGER NOT NULL REFERENCES snd_file_chunks ON DELETE CASCADE,
-  replica_number INTEGER NOT NULL,
-  xftp_server_id INTEGER NOT NULL REFERENCES xftp_servers ON DELETE CASCADE,
-  replica_id BLOB NOT NULL,
-  replica_key BLOB NOT NULL,
-  replica_status TEXT NOT NULL,
-  delay INTEGER,
-  retries INTEGER NOT NULL DEFAULT 0,
-  created_at TEXT NOT NULL DEFAULT(datetime('now')),
-  updated_at TEXT NOT NULL DEFAULT(datetime('now'))
-);
 CREATE INDEX idx_snd_file_chunk_replicas_snd_file_chunk_id ON snd_file_chunk_replicas(
  snd_file_chunk_id
 );
 CREATE INDEX idx_snd_file_chunk_replicas_xftp_server_id ON snd_file_chunk_replicas(
  xftp_server_id
 );
-CREATE TABLE snd_file_chunk_replica_recipients(
-  snd_file_chunk_replica_recipient_id INTEGER PRIMARY KEY,
-  snd_file_chunk_replica_id INTEGER NOT NULL REFERENCES snd_file_chunk_replicas ON DELETE CASCADE,
-  rcv_replica_id BLOB NOT NULL,
-  rcv_replica_key BLOB NOT NULL,
-  created_at TEXT NOT NULL DEFAULT(datetime('now')),
-  updated_at TEXT NOT NULL DEFAULT(datetime('now'))
-);
 CREATE INDEX idx_snd_file_chunk_replica_recipients_snd_file_chunk_replica_id ON snd_file_chunk_replica_recipients(
  snd_file_chunk_replica_id
 );
-CREATE TABLE deleted_snd_chunk_replicas(
-  deleted_snd_chunk_replica_id INTEGER PRIMARY KEY,
-  user_id INTEGER NOT NULL REFERENCES users ON DELETE CASCADE,
-  xftp_server_id INTEGER NOT NULL REFERENCES xftp_servers ON DELETE CASCADE,
-  replica_id BLOB NOT NULL,
-  replica_key BLOB NOT NULL,
-  chunk_digest BLOB NOT NULL,
-  delay INTEGER,
-  retries INTEGER NOT NULL DEFAULT 0,
-  created_at TEXT NOT NULL DEFAULT(datetime('now')),
-  updated_at TEXT NOT NULL DEFAULT(datetime('now'))
-);
 CREATE INDEX idx_deleted_snd_chunk_replicas_user_id ON deleted_snd_chunk_replicas(
  user_id
 );
 CREATE INDEX idx_deleted_snd_chunk_replicas_xftp_server_id ON deleted_snd_chunk_replicas(
  xftp_server_id
 );
+CREATE INDEX idx_rcv_file_chunk_replicas_pending ON rcv_file_chunk_replicas(
+  received,
+  replica_number
+);
+CREATE INDEX idx_snd_file_chunk_replicas_pending ON snd_file_chunk_replicas(
+  replica_status,
+  replica_number
+);
+CREATE INDEX idx_deleted_snd_chunk_replicas_pending ON deleted_snd_chunk_replicas(
+  created_at
+);
+CREATE INDEX idx_encrypted_rcv_message_hashes_hash ON encrypted_rcv_message_hashes(
+  conn_id,
+  hash
+);
@@ -28,7 +28,7 @@ import System.IO (BufferMode (..), hSetBuffering, stderr, stdout)
 import Text.Read (readMaybe)

 ntfServerVersion :: String
-ntfServerVersion = "1.3.0"
+ntfServerVersion = "1.4.0"

 ntfServerCLI :: FilePath -> FilePath -> IO ()
 ntfServerCLI cfgPath logPath =
@@ -741,7 +741,8 @@ restoreServerMessages = asks (storeMsgsFile . config) >>= mapM_ restoreMessages
      st <- asks queueStore
      ms <- asks msgStore
      quota <- asks $ msgQueueQuota . config
-      runExceptT (liftIO (B.readFile f) >>= mapM_ (restoreMsg st ms quota) . B.lines) >>= \case
+      old_ <- asks (messageExpiration . config) $>>= (liftIO . fmap Just . expireBeforeEpoch)
+      runExceptT (liftIO (B.readFile f) >>= mapM_ (restoreMsg st ms quota old_) . B.lines) >>= \case
        Left e -> do
          logError . T.pack $ "error restoring messages: " <> e
          liftIO exitFailure
@@ -749,7 +750,7 @@ restoreServerMessages = asks (storeMsgsFile . config) >>= mapM_ restoreMessages
          renameFile f $ f <> ".bak"
          logInfo "messages restored"
      where
-        restoreMsg st ms quota s = do
+        restoreMsg st ms quota old_ s = do
          r <- liftEither . first (msgErr "parsing") $ strDecode s
          case r of
            MLRv3 rId msg -> addToMsgQueue rId msg
@@ -759,13 +760,14 @@ restoreServerMessages = asks (storeMsgsFile . config) >>= mapM_ restoreMessages
              addToMsgQueue rId msg'
          where
            addToMsgQueue rId msg = do
-              full <- atomically $ do
+              logFull <- atomically $ do
                q <- getMsgQueue ms rId quota
-                isNothing <$> writeMsg q msg
-              case msg of
-                Message {} ->
-                  when full . logError . decodeLatin1 $ "message queue " <> strEncode rId <> " is full, message not restored: " <> strEncode (msgId (msg :: Message))
-                MessageQuota {} -> pure ()
+                case msg of
+                  Message {msgTs}
+                    | maybe True (systemSeconds msgTs >=) old_ -> isNothing <$> writeMsg q msg
+                    | otherwise -> pure False
+                  MessageQuota {} -> writeMsg q msg $> False
+              when logFull . logError . decodeLatin1 $ "message queue " <> strEncode rId <> " is full, message not restored: " <> strEncode (msgId (msg :: Message))
            updateMsgV1toV3 QueueRec {rcvDhSecret} RcvMessage {msgId, msgTs, msgFlags, msgBody = EncRcvMsgBody body} = do
              let nonce = C.cbNonce msgId
              msgBody <- liftEither . first (msgErr "v1 message decryption") $ C.maxLenBS =<< C.cbDecrypt rcvDhSecret nonce body
@@ -164,6 +164,9 @@ strictIni section key ini =
 readStrictIni :: Read a => Text -> Text -> Ini -> a
 readStrictIni section key = read . T.unpack . strictIni section key

+readIniDefault :: Read a => a -> Text -> Text -> Ini -> a
+readIniDefault def section key = either (const def) (read . T.unpack) . lookupValue section key
+
 iniOnOff :: Text -> Text -> Ini -> Maybe Bool
 iniOnOff section name ini = case lookupValue section name ini of
  Right "on" -> Just True
@@ -72,10 +72,13 @@ data ServerConfig = ServerConfig
    logTLSErrors :: Bool
  }

+defMsgExpirationDays :: Int64
+defMsgExpirationDays = 21
+
 defaultMessageExpiration :: ExpirationConfig
 defaultMessageExpiration =
  ExpirationConfig
-    { ttl = 30 * 86400, -- seconds, 30 days
+    { ttl = defMsgExpirationDays * 86400, -- seconds
      checkInterval = 43200 -- seconds, 12 hours
    }

@@ -15,3 +15,14 @@ data ExpirationConfig = ExpirationConfig

 expireBeforeEpoch :: ExpirationConfig -> IO Int64
 expireBeforeEpoch ExpirationConfig {ttl} = subtract ttl . systemSeconds <$> liftIO getSystemTime
+
+showTTL :: Int64 -> String
+showTTL s
+  | s' /= 0 = show s <> " seconds"
+  | ms' /= 0 = show ms <> " minutes"
+  | hs' /= 0 = show hs <> " hours"
+  | otherwise = show ds <> " days"
+  where
+    (ms, s') = s `divMod` 60
+    (hs, ms') = ms `divMod` 60
+    (ds, hs') = hs `divMod` 24
@@ -24,7 +24,7 @@ import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol (BasicAuth (..), ProtoServerWithAuth (ProtoServerWithAuth), pattern SMPServer)
 import Simplex.Messaging.Server (runSMPServer)
 import Simplex.Messaging.Server.CLI
-import Simplex.Messaging.Server.Env.STM (ServerConfig (..), defaultInactiveClientExpiration, defaultMessageExpiration)
+import Simplex.Messaging.Server.Env.STM (ServerConfig (..), defaultInactiveClientExpiration, defaultMessageExpiration, defMsgExpirationDays)
 import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Transport (simplexMQVersion, supportedSMPServerVRange)
 import Simplex.Messaging.Transport.Client (TransportHost (..))
@@ -106,7 +106,8 @@ smpServerCLI cfgPath logPath =
                <> ("enable: " <> onOff enableStoreLog <> "\n\n")
                <> "# Undelivered messages are optionally saved and restored when the server restarts,\n\
                   \# they are preserved in the .bak file until the next restart.\n"
-                <> ("restore_messages: " <> onOff enableStoreLog <> "\n\n")
+                <> ("restore_messages: " <> onOff enableStoreLog <> "\n")
+                <> ("expire_messages_days: " <> show defMsgExpirationDays <> "\n\n")
                <> "# Log daily server statistics to CSV file\n"
                <> ("log_stats: " <> onOff logStats <> "\n\n")
                <> "[AUTH]\n\
@@ -140,10 +141,13 @@ smpServerCLI cfgPath logPath =
      fp <- checkSavedFingerprint cfgPath defaultX509Config
      let host = fromRight "<hostnames>" $ T.unpack <$> lookupValue "TRANSPORT" "host" ini
          port = T.unpack $ strictIni "TRANSPORT" "port" ini
-          cfg@ServerConfig {transports, storeLogFile, newQueueBasicAuth, inactiveClientExpiration} = serverConfig
+          cfg@ServerConfig {transports, storeLogFile, newQueueBasicAuth, messageExpiration, inactiveClientExpiration} = serverConfig
          srv = ProtoServerWithAuth (SMPServer [THDomainName host] (if port == "5223" then "" else port) (C.KeyHash fp)) newQueueBasicAuth
      printServiceInfo serverVersion srv
      printServerConfig transports storeLogFile
+      putStrLn $ case messageExpiration of
+        Just ExpirationConfig {ttl} -> "expiring messages after " <> showTTL ttl
+        _ -> "not expiring messages"
      putStrLn $ case inactiveClientExpiration of
        Just ExpirationConfig {ttl, checkInterval} -> "expiring clients inactive for " <> show ttl <> " seconds every " <> show checkInterval <> " seconds"
        _ -> "not expiring inactive clients"
@@ -161,7 +165,7 @@ smpServerCLI cfgPath logPath =
          ServerConfig
            { transports = iniTransports ini,
              tbqSize = 32,
-              serverTbqSize = 128,
+              serverTbqSize = 1024,
              msgQueueQuota = 128,
              queueIdBytes = 24,
              msgIdBytes = 24, -- must be at least 24 bytes, it is used as 192-bit nonce for XSalsa20
@@ -179,7 +183,10 @@ smpServerCLI cfgPath logPath =
              -- allow creating new queues by default
              allowNewQueues = fromMaybe True $ iniOnOff "AUTH" "new_queues" ini,
              newQueueBasicAuth = either error id <$> strDecodeIni "AUTH" "create_password" ini,
-              messageExpiration = Just defaultMessageExpiration,
+              messageExpiration =
+                Just defaultMessageExpiration
+                  { ttl = 86400 * readIniDefault defMsgExpirationDays "STORE_LOG" "expire_messages_days" ini
+                  },
              inactiveClientExpiration =
                settingIsOn "INACTIVE_CLIENTS" "disconnect" ini
                  $> ExpirationConfig
@@ -12,13 +12,11 @@ import Control.Monad.Trans.Except
 import Data.Bifunctor (first)
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
-import Data.Fixed (Fixed (MkFixed), Pico)
 import Data.Int (Int64)
 import Data.Text (Text)
 import qualified Data.Text as T
 import Data.Text.Encoding (decodeUtf8With)
-import Data.Time (nominalDiffTimeToSeconds)
-import Data.Time.Clock (UTCTime, diffUTCTime)
+import Data.Time (NominalDiffTime, nominalDiffTimeToSeconds)
 import UnliftIO.Async

 raceAny_ :: MonadUnliftIO m => [m a] -> m ()
@@ -117,14 +115,8 @@ threadDelay' time = do
  threadDelay $ fromIntegral maxWait
  when (maxWait /= time) $ threadDelay' (time - maxWait)

-diffInSeconds :: UTCTime -> UTCTime -> Int64
-diffInSeconds a b = (`div` 1000000_000000) $ diffInPicos a b
+diffToMicroseconds :: NominalDiffTime -> Int64
+diffToMicroseconds diff = fromIntegral ((truncate $ diff * 1000000) :: Integer)

-diffInMicros :: UTCTime -> UTCTime -> Int64
-diffInMicros a b = (`div` 1000000) $ diffInPicos a b
-
-diffInPicos :: UTCTime -> UTCTime -> Int64
-diffInPicos a b = fromInteger . fromPico . nominalDiffTimeToSeconds $ diffUTCTime a b
-
-fromPico :: Pico -> Integer
-fromPico (MkFixed i) = i
+diffToMilliseconds :: NominalDiffTime -> Int64
+diffToMilliseconds diff = fromIntegral ((truncate $ diff * 1000) :: Integer)
@@ -61,6 +61,7 @@ import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Transport (ATransport (..))
 import Simplex.Messaging.Util (tryError)
 import Simplex.Messaging.Version
+import System.Directory (copyFile, renameFile)
 import Test.Hspec
 import UnliftIO
 import XFTPClient (testXFTPServer)
@@ -147,9 +148,13 @@ functionalAPITests t = do
      testAsyncServerOffline t
    it "should notify after HELLO timeout" $
      withSmpServer t testAsyncHelloTimeout
-  describe "Duplicate message delivery" $
+  describe "Message delivery" $ do
    it "should deliver messages to the user once, even if repeat delivery is made by the server (no ACK)" $
      testDuplicateMessage t
+    it "should report error via msg integrity on skipped messages" $
+      testSkippedMessages t
+    it "should report decryption error on ratchet becoming out of sync" $
+      testDecryptionError t
  describe "Inactive client disconnection" $ do
    it "should disconnect clients if it was inactive longer than TTL" $
      testInactiveClientDisconnected t
@@ -163,8 +168,10 @@ functionalAPITests t = do
    it "should suspend agent on timeout, even if pending messages not sent" $
      testSuspendingAgentTimeout t
  describe "Batching SMP commands" $ do
-    it "should subscribe to multiple subscriptions with batching" $
-      testBatchedSubscriptions t
+    xit "should subscribe to multiple (200) subscriptions with batching" $
+      testBatchedSubscriptions 200 10 t
+    it "should subscribe to multiple (6) subscriptions with batching" $
+      testBatchedSubscriptions 6 3 t
  describe "Async agent commands" $ do
    it "should connect using async agent commands" $
      withSmpServer t testAsyncCommands
@@ -491,6 +498,101 @@ testDuplicateMessage t = do
      get alice2 ##> ("", bobId, SENT 6)
      get bob2 =##> \case ("", c, Msg "hello 3") -> c == aliceId; _ -> False

+testSkippedMessages :: HasCallStack => ATransport -> IO ()
+testSkippedMessages t = do
+  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
+  bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  (aliceId, bobId) <- withSmpServerStoreLogOn t testPort $ \_ -> do
+    (aliceId, bobId) <- runRight $ makeConnection alice bob
+    runRight_ $ do
+      4 <- sendMessage alice bobId SMP.noMsgFlags "hello"
+      get alice ##> ("", bobId, SENT 4)
+      get bob =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False
+      ackMessage bob aliceId 4
+
+    disconnectAgentClient bob
+
+    runRight_ $ do
+      5 <- sendMessage alice bobId SMP.noMsgFlags "hello 2"
+      get alice ##> ("", bobId, SENT 5)
+      6 <- sendMessage alice bobId SMP.noMsgFlags "hello 3"
+      get alice ##> ("", bobId, SENT 6)
+      7 <- sendMessage alice bobId SMP.noMsgFlags "hello 4"
+      get alice ##> ("", bobId, SENT 7)
+
+    pure (aliceId, bobId)
+
+  nGet alice =##> \case ("", "", DOWN _ [c]) -> c == bobId; _ -> False
+  threadDelay 200000
+
+  disconnectAgentClient alice
+
+  alice2 <- getSMPAgentClient' agentCfg initAgentServers testDB
+  bob2 <- getSMPAgentClient' agentCfg initAgentServers testDB2
+
+  withSmpServerStoreLogOn t testPort $ \_ -> do
+    runRight_ $ do
+      subscribeConnection bob2 aliceId
+      subscribeConnection alice2 bobId
+
+      8 <- sendMessage alice2 bobId SMP.noMsgFlags "hello 5"
+      get alice2 ##> ("", bobId, SENT 8)
+      get bob2 =##> \case ("", c, MSG MsgMeta {integrity = MsgError {errorInfo = MsgSkipped {fromMsgId = 4, toMsgId = 6}}} _ "hello 5") -> c == aliceId; _ -> False
+      ackMessage bob2 aliceId 5
+
+      9 <- sendMessage alice2 bobId SMP.noMsgFlags "hello 6"
+      get alice2 ##> ("", bobId, SENT 9)
+      get bob2 =##> \case ("", c, Msg "hello 6") -> c == aliceId; _ -> False
+      ackMessage bob2 aliceId 6
+
+testDecryptionError :: HasCallStack => ATransport -> IO ()
+testDecryptionError t = do
+  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
+  bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  withSmpServerStoreMsgLogOn t testPort $ \_ -> do
+    (aliceId, bobId) <- runRight $ makeConnection alice bob
+    runRight_ $ do
+      4 <- sendMessage alice bobId SMP.noMsgFlags "hello"
+      get alice ##> ("", bobId, SENT 4)
+      get bob =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False
+      ackMessage bob aliceId 4
+
+      5 <- sendMessage bob aliceId SMP.noMsgFlags "hello 2"
+      get bob ##> ("", aliceId, SENT 5)
+      get alice =##> \case ("", c, Msg "hello 2") -> c == bobId; _ -> False
+      ackMessage alice bobId 5
+
+      liftIO $ copyFile testDB2 (testDB2 <> ".bak")
+
+      6 <- sendMessage alice bobId SMP.noMsgFlags "hello 3"
+      get alice ##> ("", bobId, SENT 6)
+      get bob =##> \case ("", c, Msg "hello 3") -> c == aliceId; _ -> False
+      ackMessage bob aliceId 6
+
+      7 <- sendMessage bob aliceId SMP.noMsgFlags "hello 4"
+      get bob ##> ("", aliceId, SENT 7)
+      get alice =##> \case ("", c, Msg "hello 4") -> c == bobId; _ -> False
+      ackMessage alice bobId 7
+
+    disconnectAgentClient bob
+
+    -- importing database backup after progressing ratchet de-synchronizes ratchet,
+    -- this will be fixed by ratchet re-negotiation
+    liftIO $ renameFile (testDB2 <> ".bak") testDB2
+
+    bob2 <- getSMPAgentClient' agentCfg initAgentServers testDB2
+
+    runRight_ $ do
+      subscribeConnection bob2 aliceId
+
+      8 <- sendMessage alice bobId SMP.noMsgFlags "hello 5"
+      get alice ##> ("", bobId, SENT 8)
+      get bob2 =##> \case ("", c, ERR AGENT {agentErr = A_CRYPTO {cryptoErr = RATCHET_HEADER}}) -> c == aliceId; _ -> False
+
+      6 <- sendMessage bob2 aliceId SMP.noMsgFlags "hello 6"
+      get bob2 ##> ("", aliceId, SENT 6)
+      get alice =##> \case ("", c, ERR AGENT {agentErr = A_CRYPTO {cryptoErr = RATCHET_HEADER}}) -> c == bobId; _ -> False
+
 makeConnection :: AgentClient -> AgentClient -> ExceptT AgentErrorType IO (ConnId, ConnId)
 makeConnection alice bob = makeConnectionForUsers alice 1 bob 1

@@ -612,14 +714,14 @@ testSuspendingAgentTimeout t = do
    ("", "", SUSPENDED) <- nGet b
    pure ()

-testBatchedSubscriptions :: ATransport -> IO ()
-testBatchedSubscriptions t = do
+testBatchedSubscriptions :: Int -> Int -> ATransport -> IO ()
+testBatchedSubscriptions nCreate nDel t = do
  a <- getSMPAgentClient' agentCfg initAgentServers2 testDB
  b <- getSMPAgentClient' agentCfg initAgentServers2 testDB2
  conns <- runServers $ do
-    conns <- forM [1 .. 200 :: Int] . const $ makeConnection a b
+    conns <- forM [1 .. nCreate :: Int] . const $ makeConnection a b
    forM_ conns $ \(aId, bId) -> exchangeGreetings a bId b aId
-    let (aIds', bIds') = unzip $ take 10 conns
+    let (aIds', bIds') = unzip $ take nDel conns
    delete a bIds'
    delete b aIds'
    liftIO $ threadDelay 1000000
@@ -635,11 +737,14 @@ testBatchedSubscriptions t = do
    ("", "", UP {}) <- nGet b
    liftIO $ threadDelay 1000000
    let (aIds, bIds) = unzip conns
-        conns' = drop 10 conns
+        conns' = drop nDel conns
        (aIds', bIds') = unzip conns'
    subscribe a bIds
    subscribe b aIds
    forM_ conns' $ \(aId, bId) -> exchangeGreetingsMsgId 6 a bId b aId
+    void $ resubscribeConnections a bIds
+    void $ resubscribeConnections b aIds
+    forM_ conns' $ \(aId, bId) -> exchangeGreetingsMsgId 8 a bId b aId
    delete a bIds'
    delete b aIds'
    deleteFail a bIds'
@@ -649,7 +754,7 @@ testBatchedSubscriptions t = do
    subscribe c cs = do
      r <- subscribeConnections c cs
      liftIO $ do
-        let dc = S.fromList $ take 10 cs
+        let dc = S.fromList $ take nDel cs
        all isRight (M.withoutKeys r dc) `shouldBe` True
        all (== Left (CONN NOT_FOUND)) (M.restrictKeys r dc) `shouldBe` True
        M.keys r `shouldMatchList` cs
@@ -428,7 +428,8 @@ mkRcvMsgData internalId internalRcvId externalSndId brokerId internalHash =
      msgFlags = SMP.noMsgFlags,
      msgBody = hw,
      internalHash,
-      externalPrevSndHash = "hash_from_sender"
+      externalPrevSndHash = "hash_from_sender",
+      encryptedMsgHash = "encrypted_msg_hash"
    }

 testCreateRcvMsg_ :: DB.Connection -> PrevExternalSndId -> PrevRcvMsgHash -> ConnId -> RcvQueue -> RcvMsgData -> Expectation
@@ -57,6 +57,8 @@ testSchemaMigrations = do
      schema'' <- getSchema testDB testSchema
      schema'' `shouldBe` schema
      withConnection st (`Migrations.run` MTRUp [m])
+      schema''' <- getSchema testDB testSchema
+      schema''' `shouldBe` schema'

 getSchema :: FilePath -> FilePath -> IO String
 getSchema dpPath schemaPath = do
@@ -55,7 +55,9 @@ serverTests t@(ATransport t') = do
    describe "Exceeding queue quota" $ testExceedQueueQuota t'
  describe "Store log" $ testWithStoreLog t
  describe "Restore messages" $ testRestoreMessages t
-  describe "Restore messages (old / v2)" $ testRestoreMessagesV2 t
+  describe "Restore messages (old / v2)" $ do
+    testRestoreMessagesV2 t
+    testRestoreExpireMessages t
  describe "Timing of AUTH error" $ testTiming t
  describe "Message notifications" $ testMessageNotifications t
  describe "Message expiration" $ do
@@ -779,6 +781,62 @@ testRestoreMessagesV2 at@(ATransport t) =
    runClient :: Transport c => TProxy c -> (THandle c -> IO ()) -> Expectation
    runClient _ test' = testSMPClient test' `shouldReturn` ()

+testRestoreExpireMessages :: ATransport -> Spec
+testRestoreExpireMessages at@(ATransport t) =
+  it "should store messages on exit and restore on start" $ do
+    (sPub, sKey) <- C.generateSignatureKeyPair C.SEd25519
+    recipientId <- newTVarIO ""
+    recipientKey <- newTVarIO Nothing
+    dhShared <- newTVarIO Nothing
+    senderId <- newTVarIO ""
+
+    withSmpServerStoreMsgLogOnV2 at testPort . runTest t $ \h -> do
+      runClient t $ \h1 -> do
+        (sId, rId, rKey, dh) <- createAndSecureQueue h1 sPub
+        atomically $ do
+          writeTVar recipientId rId
+          writeTVar recipientKey $ Just rKey
+          writeTVar dhShared $ Just dh
+          writeTVar senderId sId
+      sId <- readTVarIO senderId
+      Resp "1" _ OK <- signSendRecv h sKey ("1", sId, _SEND "hello 1")
+      Resp "2" _ OK <- signSendRecv h sKey ("2", sId, _SEND "hello 2")
+      threadDelay 3000000
+      Resp "3" _ OK <- signSendRecv h sKey ("3", sId, _SEND "hello 3")
+      Resp "4" _ OK <- signSendRecv h sKey ("4", sId, _SEND "hello 4")
+      pure ()
+
+    logSize testStoreLogFile `shouldReturn` 2
+    msgs <- B.readFile testStoreMsgsFile
+    length (B.lines msgs) `shouldBe` 4
+
+    let expCfg1 = Just ExpirationConfig {ttl = 86400, checkInterval = 43200}
+        cfg1 = cfgV2 {storeLogFile = Just testStoreLogFile, storeMsgsFile = Just testStoreMsgsFile, messageExpiration = expCfg1}
+    withSmpServerConfigOn at cfg1 testPort . runTest t $ \_ -> pure ()
+
+    logSize testStoreLogFile `shouldReturn` 1
+    msgs' <- B.readFile testStoreMsgsFile
+    msgs' `shouldBe` msgs
+
+    let expCfg2 = Just ExpirationConfig {ttl = 2, checkInterval = 43200}
+        cfg2 = cfgV2 {storeLogFile = Just testStoreLogFile, storeMsgsFile = Just testStoreMsgsFile, messageExpiration = expCfg2}
+    withSmpServerConfigOn at cfg2 testPort . runTest t $ \_ -> pure ()
+
+    logSize testStoreLogFile `shouldReturn` 1
+    -- two messages expired
+    msgs'' <- B.readFile testStoreMsgsFile
+    length (B.lines msgs'') `shouldBe` 2
+    B.lines msgs'' `shouldBe` drop 2 (B.lines msgs)
+
+  where
+    runTest :: Transport c => TProxy c -> (THandle c -> IO ()) -> ThreadId -> Expectation
+    runTest _ test' server = do
+      testSMPClient test' `shouldReturn` ()
+      killThread server
+
+    runClient :: Transport c => TProxy c -> (THandle c -> IO ()) -> Expectation
+    runClient _ test' = testSMPClient test' `shouldReturn` ()
+
 createAndSecureQueue :: Transport c => THandle c -> SndPublicVerifyKey -> IO (SenderId, RecipientId, RcvPrivateSignKey, RcvDhSecret)
 createAndSecureQueue h sPub = do
  (rPub, rKey) <- C.generateSignatureKeyPair C.SEd448
@@ -86,7 +86,7 @@ testXFTPAgentSendReceive = withXFTPServer $ do
  sndr <- getSMPAgentClient' agentCfg initAgentServers testDB
  (rfd1, rfd2) <- runRight $ do
    (sfId, _, rfd1, rfd2) <- testSend sndr filePath
-    xftpDeleteSndFileInternal sndr 1 sfId
+    xftpDeleteSndFileInternal sndr sfId
    pure (rfd1, rfd2)

  -- receive file, delete rcv file
@@ -97,7 +97,7 @@ testXFTPAgentSendReceive = withXFTPServer $ do
      rcp <- getSMPAgentClient' agentCfg initAgentServers testDB
      runRight_ $ do
        rfId <- testReceive rcp rfd originalFilePath
-        xftpDeleteRcvFile rcp 1 rfId
+        xftpDeleteRcvFile rcp rfId

 createRandomFile :: IO FilePath
 createRandomFile = do
Author	SHA1	Message	Date
spaced4ndy	cb46906aae	Merge branch 'master' into ep/test-coverage	2023-06-02 18:05:07 +04:00
spaced4ndy	b747080db3	add more message delivery tests (#763 )	2023-06-02 18:00:24 +04:00
tcely	20f7b538e5	Docker entrypoint fixes (#753 ) * docker entrypoint: DRY store log backup Don't Repeat Yourself Set a variable for the full path to the source file. Create the backup path by appending an extension from `date` output. Also fixed quoting and switched to an `if` block. Clean up block level variables. * docker entrypoint: use ISO 8601 format The previous format discards information about the local time zone. * docker entrypoint: always use UTC Now the format always ends in +00:00, and we can ignore that part again. * docker entrypoint: provide the date format again * docker entrypoint: remove time zone from the date format * docker entrypoint: use an unambiguous date format Present a leading zero before the month: YYYY-0MM-DD Both YYYY-MM-DD and YYYY-DD-MM are used by people and can be confusing in the beginning of the month. * docker entrypoint: use appropriate quoting Without avoiding field splitting a password containing a space changes the number of arguments being set. * docker entrypoint: use explicit braces for custom variables Make the intentions clear and don't assume the user knows the special cases for when variables won't be extended. Example: word=animal words=mistake echo "$words vs $word vs ${word}s" * docker entrypoint-xftp-server: braces and quoting * docker entrypoint-xftp-server: backup block * docker entrypoints: explain date format in a comment * switched from long to short option to date for POSIX * docker entrypoint-smp-server: explain date format * docker entrypoint-xftp-server: explain date format * docker entrypoints: further explain format I fixed the case to match the date format letters. Also, use words to explain since I don't want everyone to need to read about date formats to understand. * docker entrypoints: only quote letters I was either going to quote the dashes too or stop quoting the colons. Having less quotes was more readable. * Revert "docker entrypoint: use an unambiguous date format" This reverts commit `ba2a93bad9`. * docker entrypoints: remove %F * docker entrypoint-smp-server: remove %F Used part of the explicit ISO 8601 format, provided by the coreutils date invocation guide. * docker entrypoint-xftp-server: remove %F Used part of the explicit ISO 8601 format, provided by the coreutils date invocation guide.	2023-05-29 14:12:23 +03:00
Evgeny Poberezkin	1d87f8e420	test coverage	2023-05-29 09:36:37 +01:00
Evgeny Poberezkin	89caf55729	agent: ack message in case of decryption error (#759 )	2023-05-27 18:28:14 +01:00
Evgeny Poberezkin	c08cfb460b	5.1.1	2023-05-27 15:54:40 +01:00
Evgeny Poberezkinandspaced4ndy	03eca19d38	server: expire messages when restoring them (#758 ) * server: expire messages when restoring them * add test * specify constructor Co-authored-by: spaced4ndy <8711996+spaced4ndy@users.noreply.github.com> * improve test --------- Co-authored-by: spaced4ndy <8711996+spaced4ndy@users.noreply.github.com>	2023-05-27 15:52:49 +01:00
Evgeny Poberezkin	a5a3a2cbad	server: increase server main queue size	2023-05-27 15:18:53 +01:00
spaced4ndy	4a3c5abf32	add with withTransaction version with logging (#757 )	2023-05-26 20:33:48 +04:00
spaced4ndy	6c6f22051d	check previous encrypted message hashes for duplicates on decryption errors (#756 )	2023-05-17 15:42:57 +04:00
spaced4ndy	d693868bc0	change implementation of time diff calculation (#754 )	2023-05-15 20:26:18 +04:00
spaced4ndy	32a64b994e	remove agentOperationBracket from file transfer (#755 )	2023-05-15 20:23:18 +04:00
Evgeny Poberezkin	a83f85dc58	5.1.0: smp 1.4.0, xftp 1.0.0	2023-05-12 22:01:32 +01:00
sh	0f7ede5eed	docker+scripts: XFTP servers installation, update docs, install scripts (#750 ) * docker: major overhaul with xftp support * github: update docker actions * install: add installation scripts * readme: update readme to reflect latest changes	2023-05-12 10:45:04 +01:00
spaced4ndy	ce64c91d5a	add indexes for reading pending files (#751 )	2023-05-10 15:17:26 +04:00
tcely shum Evgeny Poberezkin	66177fd550	Add shumvgolove to CODEOWNERS (#747 ) * CODEOWNERS: add shumvgolove for docker scripts * CODEOWNERS: add shumvgolove for Dockerfiles * CODEOWNERS: add shumvgolove for main scripts * update --------- Co-authored-by: shum <shum@liber.li> Co-authored-by: Evgeny Poberezkin <2769109+epoberezkin@users.noreply.github.com>	2023-05-09 20:08:17 +01:00
Evgeny Poberezkinandspaced4ndy	9d83a9c017	configure message/file expiration time in INI file (#749 ) * configure message/file expiration time in INI file * correct comment Co-authored-by: spaced4ndy <8711996+spaced4ndy@users.noreply.github.com> --------- Co-authored-by: spaced4ndy <8711996+spaced4ndy@users.noreply.github.com>	2023-05-09 20:07:42 +01:00
tcely	019db0ab91	DRY build.Dockerfile (#748 ) In addition to Don't Repeat Yourself, using ENV shows the versions used in the image history.	2023-05-09 20:06:12 +01:00
spaced4ndyandEvgeny Poberezkin	8954f39425	agent: vacuum on migrations (#745 ) * agent: vacuum on migrations * vacuum after all migrations * style --------- Co-authored-by: Evgeny Poberezkin <2769109+epoberezkin@users.noreply.github.com>	2023-05-04 17:26:05 +04:00
Sam A	4d96e3700c	Add GH Actions workflow to build and push Docker image (#734 ) * Add GH Actions workflow to build and push Docker image on release * Use a matrix to support building other applications * Rename application from simplexmq to smp-server * Remove linux/386 platform * Remove linux/arm64 platform	2023-04-26 08:53:34 +01:00
sh	eaa8221b95	docker: adjusments (#737 ) * docker: workaround for cabal * docker: use llvm-12 See: https://gitlab.haskell.org/ghc/ghc/-/issues/21936 * docker: switch to jammy	2023-04-25 11:31:15 +01:00
spaced4ndy	af3f70829d	xftp: remove user id from internal deletion apis (#739 )	2023-04-25 12:48:10 +04:00
spaced4ndy	ff8197b87b	xftp: check encrypted file exists when uploading (#738 )	2023-04-24 19:45:30 +04:00
sh	fd96ee2840	docker: revert buildplatform (#736 ) * docker: revert buildplatform * docker: only build smp-server	2023-04-24 13:34:26 +01:00
sh	605970f6b6	docker: refactoring (#735 ) * build: switch ubuntu image to lunar * dcoker: support multi-platform builds * docker: strip binary * docker: multicore execution * readme: change docker hub to smp-server	2023-04-24 11:10:05 +01:00
sh	dc40c3461a	docker: fix docker logs and add build arg (#733 )	2023-04-23 15:29:19 +01:00