diff --git a/.github/workflows/docker-image.yml b/.github/workflows/docker-image.yml index e1f4c87f1..efd74552f 100644 --- a/.github/workflows/docker-image.yml +++ b/.github/workflows/docker-image.yml @@ -14,22 +14,22 @@ jobs: matrix: include: - app: smp-server - app_port: 5223 + app_port: "443 5223" - app: xftp-server - app_port: 443 + app_port: 443 steps: - name: Clone project - uses: actions/checkout@v3 + uses: actions/checkout@v4 - name: Log in to Docker Hub - uses: docker/login-action@v2 + uses: docker/login-action@v3 with: username: ${{ secrets.DOCKERHUB_USERNAME }} password: ${{ secrets.DOCKERHUB_PASSWORD }} - name: Extract metadata for Docker image id: meta - uses: docker/metadata-action@v4 + uses: docker/metadata-action@v5 with: images: ${{ secrets.DOCKERHUB_USERNAME }}/${{ matrix.app }} flavor: | @@ -40,7 +40,7 @@ jobs: type=semver,pattern=v{{major}} - name: Build and push Docker image - uses: docker/build-push-action@v4 + uses: docker/build-push-action@v6 with: push: true build-args: | diff --git a/Dockerfile b/Dockerfile index e81db17c3..06b59e6e1 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,15 +1,20 @@ -ARG TAG=22.04 +# syntax=docker/dockerfile:1.7.0-labs +ARG TAG=24.04 FROM ubuntu:${TAG} AS build ### Build stage # Install curl and git and simplexmq dependencies -RUN apt-get update && apt-get install -y curl git build-essential libgmp3-dev zlib1g-dev llvm-12 llvm-12-dev libnuma-dev libssl-dev +RUN apt-get update && apt-get install -y curl git build-essential libgmp3-dev zlib1g-dev llvm-18 llvm-18-dev libnuma-dev libssl-dev # Specify bootstrap Haskell versions ENV BOOTSTRAP_HASKELL_GHC_VERSION=9.6.3 -ENV BOOTSTRAP_HASKELL_CABAL_VERSION=3.10.1.0 +ENV BOOTSTRAP_HASKELL_CABAL_VERSION=3.12.1.0 + +# Do not install Stack +ENV BOOTSTRAP_HASKELL_INSTALL_NO_STACK=true +ENV BOOTSTRAP_HASKELL_INSTALL_NO_STACK_HOOK=true # Install ghcup RUN curl --proto '=https' --tlsv1.2 -sSf https://get-ghcup.haskell.org | BOOTSTRAP_HASKELL_NONINTERACTIVE=1 sh @@ -21,26 +26,42 @@ ENV PATH="/root/.cabal/bin:/root/.ghcup/bin:$PATH" RUN ghcup set ghc "${BOOTSTRAP_HASKELL_GHC_VERSION}" && \ ghcup set cabal "${BOOTSTRAP_HASKELL_CABAL_VERSION}" -COPY . /project +# Copy only the source code +COPY apps /project/apps/ +COPY cbits /project/cbits/ +COPY src /project/src/ + +COPY cabal.project Setup.hs simplexmq.cabal LICENSE /project + WORKDIR /project +# Debug +#ARG CACHEBUST=1 + +#ADD --chmod=755 https://github.com/MShekow/directory-checksum/releases/download/v1.4.6/directory-checksum_1.4.6_linux_amd64 /usr/local/bin/directory-checksum +#RUN directory-checksum --max-depth 2 . + +# Set build arguments and check if they exist ARG APP -ARG APP_PORT -RUN if [ -z "$APP" ] || [ -z "$APP_PORT" ]; then printf "Please spcify \$APP and \$APP_PORT build-arg.\n"; exit 1; fi +RUN if [ -z "$APP" ]; then printf "Please spcify \$APP build-arg.\n"; exit 1; fi # Compile app RUN cabal update RUN cabal build exe:$APP +# Copy scripts +COPY scripts /project/scripts/ + # Create new path containing all files needed RUN mkdir /final WORKDIR /final # Strip the binary from debug symbols to reduce size -RUN bin=$(find /project/dist-newstyle -name "$APP" -type f -executable) && \ +RUN bin="$(find /project/dist-newstyle -name "$APP" -type f -executable)" && \ mv "$bin" ./ && \ strip ./"$APP" &&\ - mv /project/scripts/docker/entrypoint-"$APP" ./entrypoint + mv /project/scripts/docker/entrypoint-"$APP" ./entrypoint &&\ + mv /project/scripts/main/simplex-servers-stopscript ./simplex-servers-stopscript ### Final stage FROM ubuntu:${TAG} @@ -53,6 +74,8 @@ COPY --from=build /final /usr/local/bin/ # Open app listening port ARG APP_PORT +RUN if [ -z "$APP_PORT" ]; then printf "Please spcify \$APP_PORT build-arg.\n"; exit 1; fi + EXPOSE $APP_PORT # simplexmq requires using SIGINT to correctly preserve undelivered messages and restore them on restart diff --git a/scripts/docker/docker-compose-smp-complete.yml b/scripts/docker/docker-compose-smp-complete.yml new file mode 100644 index 000000000..be871983a --- /dev/null +++ b/scripts/docker/docker-compose-smp-complete.yml @@ -0,0 +1,67 @@ +name: SimpleX Chat - smp-server + +services: + oneshot: + image: ubuntu:latest + environment: + CADDYCONF: | + ${CADDY_OPTS:-} + + http://{$$ADDR} { + redir https://{$$ADDR}{uri} permanent + } + + {$$ADDR}:8443 { + tls { + key_type rsa4096 + } + } + command: sh -c 'if [ ! -f /etc/caddy/Caddyfile ]; then printf "$${CADDYCONF}" > /etc/caddy/Caddyfile; fi' + volumes: + - ./caddy_conf:/etc/caddy + + caddy: + image: caddy:latest + depends_on: + oneshot: + condition: service_completed_successfully + cap_add: + - NET_ADMIN + environment: + ADDR: ${ADDR?"Please specify the domain."} + volumes: + - ./caddy_conf:/etc/caddy + - caddy_data:/data + - caddy_config:/config + ports: + - 80:80 + restart: unless-stopped + healthcheck: + test: "test -d /data/caddy/certificates/${CERT_PATH:-acme-v02.api.letsencrypt.org-directory}/${ADDR} || exit 1" + interval: 1s + retries: 60 + + smp-server: + image: ${SIMPLEX_IMAGE:-simplexchat/smp-server:latest} + depends_on: + caddy: + condition: service_healthy + environment: + ADDR: ${ADDR?"Please specify the domain."} + PASS: ${PASS:-} + volumes: + - ./smp_configs:/etc/opt/simplex + - ./smp_state:/var/opt/simplex + - type: volume + source: caddy_data + target: /certificates + volume: + subpath: "caddy/certificates/${CERT_PATH:-acme-v02.api.letsencrypt.org-directory}/${ADDR}" + ports: + - 443:443 + - 5223:5223 + restart: unless-stopped + +volumes: + caddy_data: + caddy_config: diff --git a/scripts/docker/docker-compose-smp-manual.yml b/scripts/docker/docker-compose-smp-manual.yml new file mode 100644 index 000000000..391219b15 --- /dev/null +++ b/scripts/docker/docker-compose-smp-manual.yml @@ -0,0 +1,15 @@ +name: SimpleX Chat - smp-server + +services: + smp-server: + image: ${SIMPLEX_IMAGE:-simplexchat/smp-server:latest} + environment: + WEB_MANUAL: ${WEB_MANUAL:-1} + ADDR: ${ADDR?"Please specify the domain."} + PASS: ${PASS:-} + volumes: + - ./smp_configs:/etc/opt/simplex + - ./smp_state:/var/opt/simplex + ports: + - 5223:5223 + restart: unless-stopped diff --git a/scripts/docker/docker-compose-smp.env b/scripts/docker/docker-compose-smp.env new file mode 100644 index 000000000..17c51cf9a --- /dev/null +++ b/scripts/docker/docker-compose-smp.env @@ -0,0 +1,11 @@ +# Mandatory +ADDR=your_ip_or_addr + +# Optional +#PASS='123123' +#WEB_MANUAL=1 + +# Debug +#SIMPLEX_SMP_IMAGE=smp-server-dev +#CERT_PATH=acme-staging-v02.api.letsencrypt.org-directory +#CADDY_OPTS='{\n acme_ca https://acme-staging-v02.api.letsencrypt.org/directory\n}' diff --git a/scripts/docker/docker-compose-xftp.env b/scripts/docker/docker-compose-xftp.env new file mode 100644 index 000000000..425b769a3 --- /dev/null +++ b/scripts/docker/docker-compose-xftp.env @@ -0,0 +1,9 @@ +# Mandatory +ADDR=your_ip_or_addr +QUOTA=120gb + +# Optional +#PASS='123123' + +# Debug +#SIMPLEX_XFTP_IMAGE=xftp-server-dev diff --git a/scripts/docker/docker-compose-xftp.yml b/scripts/docker/docker-compose-xftp.yml new file mode 100644 index 000000000..0686412f3 --- /dev/null +++ b/scripts/docker/docker-compose-xftp.yml @@ -0,0 +1,16 @@ +name: SimpleX Chat - xftp-server + +services: + xftp-server: + image: ${SIMPLEX_XFTP_IMAGE:-simplexchat/xftp-server:latest} + environment: + ADDR: ${ADDR?"Please specify the domain."} + QUOTA: ${QUOTA?"Please specify disk quota."} + PASS: ${PASS:-} + volumes: + - ./xftp_configs:/etc/opt/simplex-xftp + - ./xftp_state:/var/opt/simplex-xftp + - ./xftp_files:/srv/xftp + ports: + - 443:443 + restart: unless-stopped diff --git a/scripts/docker/entrypoint-smp-server b/scripts/docker/entrypoint-smp-server index 5817b7b56..eeea3582d 100755 --- a/scripts/docker/entrypoint-smp-server +++ b/scripts/docker/entrypoint-smp-server @@ -1,48 +1,87 @@ #!/usr/bin/env sh +set -e + confd='/etc/opt/simplex' -logd='/var/opt/simplex/' +cert_path='/certificates' # Check if server has been initialized if [ ! -f "${confd}/smp-server.ini" ]; then # If not, determine ip or domain case "${ADDR}" in - '') printf 'Please specify $ADDR environment variable.\n'; exit 1 ;; + '') + printf 'Please specify $ADDR environment variable.\n' + exit 1 + ;; + + # Determine domain or IPv6 *[a-zA-Z]*) case "${ADDR}" in - *:*) set -- --ip "${ADDR}" ;; - *) set -- -n "${ADDR}" ;; + # IPv6 + *:*) + set -- --ip "${ADDR}" + ;; + + # Domain + *) + case "${ADDR}" in + # It's in domain format + *.*) + # Determine the base domain + ADDR_BASE="$(printf '%s' "$ADDR" | awk -F. '{print $(NF-1)"."$NF}')" + set -- --fqdn "${ADDR}" --own-domains="${ADDR_BASE}" + ;; + + # Incorrect domain + *) + printf 'Incorrect $ADDR environment variable. Please specify the correct one in format: smp1.example.org / example.org \n' + exit 1 + ;; + esac esac ;; - *) set -- --ip "${ADDR}" ;; + + # Assume everything else is IPv4 + *) + set -- --ip "${ADDR}" ;; esac # Optionally, set password case "${PASS}" in - '') set -- "$@" --no-password ;; - *) set -- "$@" --password "${PASS}" ;; + # Empty value = no password + '') + set -- "$@" --no-password + ;; + + # Assume that everything else is a password + *) + set -- "$@" --password "${PASS}" + ;; esac # And init certificates and configs - smp-server init -y -l "$@" + smp-server init --yes \ + --store-log \ + --daily-stats \ + --source-code \ + "$@" > /dev/null 2>&1 + + # Fix path to certificates + if [ -n "${WEB_MANUAL}" ]; then + sed -i -e 's|^[^#]*https: |#&|' \ + -e 's|^[^#]*cert: |#&|' \ + -e 's|^[^#]*key: |#&|' \ + -e 's|^port:.*|port: 5223|' \ + "${confd}/smp-server.ini" + else + sed -i -e "s|cert: /etc/opt/simplex/web.crt|cert: $cert_path/$ADDR.crt|" \ + -e "s|key: /etc/opt/simplex/web.key|key: $cert_path/$ADDR.key|" \ + "${confd}/smp-server.ini" + fi fi # Backup store log just in case -# -# Uses the UTC (universal) time zone and this -# format: YYYY-mm-dd'T'HH:MM:SS -# year, month, day, letter T, hour, minute, second -# -# This is the ISO 8601 format without the time zone at the end. -# -_file="${logd}/smp-server-store.log" -if [ -f "${_file}" ]; then - _backup_extension="$(date -u '+%Y-%m-%dT%H:%M:%S')" - cp -v -p "${_file}" "${_file}.${_backup_extension:-date-failed}" - unset -v _backup_extension -fi -unset -v _file +DOCKER=true /usr/local/bin/simplex-servers-stopscript smp-server # Finally, run smp-sever. Notice that "exec" here is important: # smp-server replaces our helper script, so that it can catch INT signal exec smp-server start +RTS -N -RTS - diff --git a/scripts/docker/entrypoint-xftp-server b/scripts/docker/entrypoint-xftp-server index 9e5bf5ac1..31d75362e 100755 --- a/scripts/docker/entrypoint-xftp-server +++ b/scripts/docker/entrypoint-xftp-server @@ -1,50 +1,90 @@ #!/usr/bin/env sh +set -eu + confd='/etc/opt/simplex-xftp' -logd='/var/opt/simplex-xftp' # Check if server has been initialized if [ ! -f "${confd}/file-server.ini" ]; then # If not, determine ip or domain case "${ADDR}" in - '') printf 'Please specify $ADDR environment variable.\n'; exit 1 ;; + '') + printf 'Please specify $ADDR environment variable.\n' + exit 1 + ;; + + # Determine domain or IPv6 *[a-zA-Z]*) case "${ADDR}" in - *:*) set -- --ip "${ADDR}" ;; - *) set -- -n "${ADDR}" ;; + # IPv6 + *:*) + set -- --ip "${ADDR}" + ;; + + # Domain + *) + case "${ADDR}" in + # Check if format is correct + *.*) + set -- --fqdn "${ADDR}" + ;; + + # Incorrect domain + *) + printf 'Incorrect $ADDR environment variable. Please specify the correct one in format: smp1.example.org / example.org \n' + exit 1 + ;; + esac + ;; esac ;; - *) set -- --ip "${ADDR}" ;; + + # Assume everything else is IPv4 + *) + set -- --ip "${ADDR}" + ;; esac - # Set quota + # Set global disk quota case "${QUOTA}" in - '') printf 'Please specify $QUOTA environment variable.\n'; exit 1 ;; - *GB) QUOTA="$(printf ${QUOTA} | tr '[:upper:]' '[:lower:]')"; set -- "$@" --quota "${QUOTA}" ;; - *gb) set -- "$@" --quota "${QUOTA}" ;; - *) printf 'Wrong format. Format should be: 1gb, 10gb, 100gb.\n'; exit 1 ;; + '') + printf 'Please specify $QUOTA environment variable.\n' + exit 1 + ;; + + # Incorrect format in uppercase, but automagically workaround this, replacing characters to lowercase + *GB) + QUOTA="$(printf '%s' "${QUOTA}" | tr '[:upper:]' '[:lower:]')" + set -- "$@" --quota "${QUOTA}" + ;; + + # Correct format + *gb) + set -- "$@" --quota "${QUOTA}" + ;; + + # Incorrect format + *) + printf 'Wrong format. Format should be: 1gb, 10gb, 100gb.\n' + exit 1 + ;; esac # Init the certificates and configs - xftp-server init -l -p /srv/xftp "$@" + xftp-server init --store-log \ + --path /srv/xftp \ + "$@" > /dev/null 2>&1 + + # Optionally, set password + if [ -n "${PASS}" ]; then + sed -i -e "/^# create_password:/a create_password: $PASS" \ + "${confd}/file-server.ini" + fi fi # Backup store log just in case -# -# Uses the UTC (universal) time zone and this -# format: YYYY-mm-dd'T'HH:MM:SS -# year, month, day, letter T, hour, minute, second -# -# This is the ISO 8601 format without the time zone at the end. -# -_file="${logd}/file-server-store.log" -if [ -f "${_file}" ]; then - _backup_extension="$(date -u '+%Y-%m-%dT%H:%M:%S')" - cp -v -p "${_file}" "${_file}.${_backup_extension:-date-failed}" - unset -v _backup_extension -fi -unset -v _file + +DOCKER=true /usr/local/bin/simplex-servers-stopscript xftp-server # Finally, run xftp-sever. Notice that "exec" here is important: # smp-server replaces our helper script, so that it can catch INT signal exec xftp-server start +RTS -N -RTS - diff --git a/scripts/main/simplex-servers-stopscript b/scripts/main/simplex-servers-stopscript index acfed3431..0b7003ead 100755 --- a/scripts/main/simplex-servers-stopscript +++ b/scripts/main/simplex-servers-stopscript @@ -148,8 +148,10 @@ xftp_cleanup() { main() { type="${1:-}" - - checks + + if [ -z "${DOCKER+x}" ]; then + checks + fi case "$type" in smp-server) diff --git a/scripts/main/smp-server.service b/scripts/main/smp-server.service index 6d365041d..61c695217 100644 --- a/scripts/main/smp-server.service +++ b/scripts/main/smp-server.service @@ -5,12 +5,21 @@ Description=SMP server User=smp Group=smp Type=simple + ExecStart=/usr/local/bin/smp-server start +RTS -N -RTS ExecStopPost=/usr/local/bin/simplex-servers-stopscript smp-server + LimitNOFILE=65535 KillSignal=SIGINT + TimeoutStartSec=infinity TimeoutStopSec=infinity + +Restart=on-failure +RestartSec=10s +StartLimitBurst=3 +StartLimitInterval=60s + AmbientCapabilities=CAP_NET_BIND_SERVICE [Install] diff --git a/scripts/main/xftp-server.service b/scripts/main/xftp-server.service index fcde29bf8..32229a47e 100644 --- a/scripts/main/xftp-server.service +++ b/scripts/main/xftp-server.service @@ -5,12 +5,21 @@ Description=XFTP server User=xftp Group=xftp Type=simple + ExecStart=/usr/local/bin/xftp-server start +RTS -N -RTS ExecStopPost=/usr/local/bin/simplex-servers-stopscript xftp-server + LimitNOFILE=65535 KillSignal=SIGINT + TimeoutStartSec=infinity TimeoutStopSec=infinity + +Restart=on-failure +RestartSec=10s +StartLimitBurst=3 +StartLimitInterval=60s + AmbientCapabilities=CAP_NET_BIND_SERVICE [Install] diff --git a/simplexmq.cabal b/simplexmq.cabal index bade74603..f849715b2 100644 --- a/simplexmq.cabal +++ b/simplexmq.cabal @@ -1,7 +1,7 @@ cabal-version: 1.12 name: simplexmq -version: 6.3.0.2 +version: 6.3.0.3 synopsis: SimpleXMQ message broker description: This package includes <./docs/Simplex-Messaging-Server.html server>, <./docs/Simplex-Messaging-Client.html client> and diff --git a/src/Simplex/FileTransfer/Description.hs b/src/Simplex/FileTransfer/Description.hs index 0c7c42ab4..cd2df9d33 100644 --- a/src/Simplex/FileTransfer/Description.hs +++ b/src/Simplex/FileTransfer/Description.hs @@ -1,4 +1,3 @@ -{-# LANGUAGE CPP #-} {-# LANGUAGE DataKinds #-} {-# LANGUAGE DeriveAnyClass #-} {-# LANGUAGE DerivingStrategies #-} @@ -71,20 +70,13 @@ import qualified Data.Yaml as Y import Simplex.FileTransfer.Chunks import Simplex.FileTransfer.Protocol import Simplex.Messaging.Agent.QueryString -import Simplex.Messaging.Agent.Store.DB (Binary (..)) +import Simplex.Messaging.Agent.Store.DB (Binary (..), FromField (..), ToField (..)) import qualified Simplex.Messaging.Crypto as C import Simplex.Messaging.Encoding.String import Simplex.Messaging.Parsers (defaultJSON, parseAll) import Simplex.Messaging.Protocol (XFTPServer) import Simplex.Messaging.ServiceScheme (ServiceScheme (..)) import Simplex.Messaging.Util (bshow, safeDecodeUtf8, (<$?>)) -#if defined(dbPostgres) -import Database.PostgreSQL.Simple.FromField (FromField (..)) -import Database.PostgreSQL.Simple.ToField (ToField (..)) -#else -import Database.SQLite.Simple.FromField (FromField (..)) -import Database.SQLite.Simple.ToField (ToField (..)) -#endif data FileDescription (p :: FileParty) = FileDescription { party :: SFileParty p, diff --git a/src/Simplex/FileTransfer/Types.hs b/src/Simplex/FileTransfer/Types.hs index c18d31779..d80ff7c77 100644 --- a/src/Simplex/FileTransfer/Types.hs +++ b/src/Simplex/FileTransfer/Types.hs @@ -1,4 +1,3 @@ -{-# LANGUAGE CPP #-} {-# LANGUAGE DuplicateRecordFields #-} {-# LANGUAGE LambdaCase #-} {-# LANGUAGE NamedFieldPuns #-} @@ -23,13 +22,7 @@ import Simplex.Messaging.Encoding.String import Simplex.Messaging.Parsers import Simplex.Messaging.Protocol (XFTPServer) import System.FilePath (()) -#if defined(dbPostgres) -import Database.PostgreSQL.Simple.FromField (FromField (..)) -import Database.PostgreSQL.Simple.ToField (ToField (..)) -#else -import Database.SQLite.Simple.FromField (FromField (..)) -import Database.SQLite.Simple.ToField (ToField (..)) -#endif +import Simplex.Messaging.Agent.Store.DB (FromField (..), ToField (..)) type RcvFileId = ByteString -- Agent entity ID diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index 709f129bd..2ff0d2ab2 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/src/Simplex/Messaging/Agent.hs @@ -1805,8 +1805,8 @@ prepareDeleteConnections_ getConnections c waitDelivery connIds = do -- ! if it was used to notify about the result, it might be necessary to differentiate -- ! between completed deletions of connections, and deletions delayed due to wait for delivery (see deleteConn) deliveryTimeout <- if waitDelivery then asks (Just . connDeleteDeliveryTimeout . config) else pure Nothing - rs' <- lift $ catMaybes . rights <$> withStoreBatch' c (\db -> map (deleteConn db deliveryTimeout) (M.keys delRs)) - forM_ rs' $ \cId -> notify ("", cId, AEvt SAEConn DEL_CONN) + cIds_ <- lift $ L.nonEmpty . catMaybes . rights <$> withStoreBatch' c (\db -> map (deleteConn db deliveryTimeout) (M.keys delRs)) + forM_ cIds_ $ \cIds -> notify ("", "", AEvt SAEConn $ DEL_CONNS cIds) pure (errs' <> delRs, rqs, connIds') where rcvQueues :: SomeConn -> Either (Either AgentErrorType ()) [RcvQueue] @@ -1826,32 +1826,33 @@ deleteConnQueues c waitDelivery ntf rqs = do rs <- connResults <$> (deleteQueueRecs =<< deleteQueues c rqs) let connIds = M.keys $ M.filter isRight rs deliveryTimeout <- if waitDelivery then asks (Just . connDeleteDeliveryTimeout . config) else pure Nothing - rs' <- catMaybes . rights <$> withStoreBatch' c (\db -> map (deleteConn db deliveryTimeout) connIds) - forM_ rs' $ \cId -> notify ("", cId, AEvt SAEConn DEL_CONN) + cIds_ <- L.nonEmpty . catMaybes . rights <$> withStoreBatch' c (\db -> map (deleteConn db deliveryTimeout) connIds) + forM_ cIds_ $ \cIds -> notify ("", "", AEvt SAEConn $ DEL_CONNS cIds) pure rs where deleteQueueRecs :: [(RcvQueue, Either AgentErrorType ())] -> AM' [(RcvQueue, Either AgentErrorType ())] deleteQueueRecs rs = do maxErrs <- asks $ deleteErrorCount . config - (rs', notifyActions) <- unzip . rights <$> withStoreBatch' c (\db -> map (deleteQueueRec db maxErrs) rs) - mapM_ sequence_ notifyActions - pure rs' + rs' <- rights <$> withStoreBatch' c (\db -> map (deleteQueueRec db maxErrs) rs) + let delQ ((rq, _), err_) = (qConnId rq,qServer rq,queueId rq,) <$> err_ + delQs_ = L.nonEmpty $ mapMaybe delQ rs' + forM_ delQs_ $ \delQs -> notify ("", "", AEvt SAEConn $ DEL_RCVQS delQs) + pure $ map fst rs' where deleteQueueRec :: DB.Connection -> Int -> (RcvQueue, Either AgentErrorType ()) -> - IO ((RcvQueue, Either AgentErrorType ()), Maybe (AM' ())) + IO ((RcvQueue, Either AgentErrorType ()), Maybe (Maybe AgentErrorType)) -- Nothing - no event, Just Nothing - no error deleteQueueRec db maxErrs (rq@RcvQueue {userId, server}, r) = case r of - Right _ -> deleteConnRcvQueue db rq $> ((rq, r), Just (notifyRQ rq Nothing)) + Right _ -> deleteConnRcvQueue db rq $> ((rq, r), Just Nothing) Left e | temporaryOrHostError e && deleteErrors rq + 1 < maxErrs -> incRcvDeleteErrors db rq $> ((rq, r), Nothing) | otherwise -> do deleteConnRcvQueue db rq -- attempts and successes are counted in deleteQueues function atomically $ incSMPServerStat c userId server connDeleted - pure ((rq, Right ()), Just (notifyRQ rq (Just e))) - notifyRQ rq e_ = notify ("", qConnId rq, AEvt SAEConn $ DEL_RCVQ (qServer rq) (queueId rq) e_) + pure ((rq, Right ()), Just (Just e)) notify = when ntf . atomically . writeTBQueue (subQ c) connResults :: [(RcvQueue, Either AgentErrorType ())] -> Map ConnId (Either AgentErrorType ()) connResults = M.map snd . foldl' addResult M.empty diff --git a/src/Simplex/Messaging/Agent/Protocol.hs b/src/Simplex/Messaging/Agent/Protocol.hs index b87f87f18..c5219ab22 100644 --- a/src/Simplex/Messaging/Agent/Protocol.hs +++ b/src/Simplex/Messaging/Agent/Protocol.hs @@ -1,4 +1,3 @@ -{-# LANGUAGE CPP #-} {-# LANGUAGE DataKinds #-} {-# LANGUAGE DeriveAnyClass #-} {-# LANGUAGE DuplicateRecordFields #-} @@ -168,12 +167,12 @@ import Data.Time.Clock.System (SystemTime) import Data.Type.Equality import Data.Typeable () import Data.Word (Word16, Word32) +import Simplex.Messaging.Agent.Store.DB (FromField (..), ToField (..)) import Simplex.FileTransfer.Description import Simplex.FileTransfer.Protocol (FileParty (..)) import Simplex.FileTransfer.Transport (XFTPErrorType) import Simplex.FileTransfer.Types (FileErrorType) import Simplex.Messaging.Agent.QueryString -import Simplex.Messaging.Agent.Store.DB (Binary (..)) import Simplex.Messaging.Client (ProxyClientError) import qualified Simplex.Messaging.Crypto as C import Simplex.Messaging.Crypto.Ratchet @@ -224,13 +223,6 @@ import Simplex.Messaging.Version import Simplex.Messaging.Version.Internal import Simplex.RemoteControl.Types import UnliftIO.Exception (Exception) -#if defined(dbPostgres) -import Database.PostgreSQL.Simple.FromField (FromField (..)) -import Database.PostgreSQL.Simple.ToField (ToField (..)) -#else -import Database.SQLite.Simple.FromField (FromField (..)) -import Database.SQLite.Simple.ToField (ToField (..)) -#endif -- SMP agent protocol version history: -- 1 - binary protocol encoding (1/1/2022) @@ -366,8 +358,8 @@ data AEvent (e :: AEntity) where MSGNTF :: MsgId -> Maybe UTCTime -> AEvent AEConn RCVD :: MsgMeta -> NonEmpty MsgReceipt -> AEvent AEConn QCONT :: AEvent AEConn - DEL_RCVQ :: SMPServer -> SMP.RecipientId -> Maybe AgentErrorType -> AEvent AEConn - DEL_CONN :: AEvent AEConn + DEL_RCVQS :: NonEmpty (ConnId, SMPServer, SMP.RecipientId, Maybe AgentErrorType) -> AEvent AEConn + DEL_CONNS :: NonEmpty ConnId -> AEvent AEConn DEL_USER :: Int64 -> AEvent AENone STAT :: ConnectionStats -> AEvent AEConn OK :: AEvent AEConn @@ -437,8 +429,8 @@ data AEventTag (e :: AEntity) where MSGNTF_ :: AEventTag AEConn RCVD_ :: AEventTag AEConn QCONT_ :: AEventTag AEConn - DEL_RCVQ_ :: AEventTag AEConn - DEL_CONN_ :: AEventTag AEConn + DEL_RCVQS_ :: AEventTag AEConn + DEL_CONNS_ :: AEventTag AEConn DEL_USER_ :: AEventTag AENone STAT_ :: AEventTag AEConn OK_ :: AEventTag AEConn @@ -492,8 +484,8 @@ aEventTag = \case MSGNTF {} -> MSGNTF_ RCVD {} -> RCVD_ QCONT -> QCONT_ - DEL_RCVQ {} -> DEL_RCVQ_ - DEL_CONN -> DEL_CONN_ + DEL_RCVQS _ -> DEL_RCVQS_ + DEL_CONNS _ -> DEL_CONNS_ DEL_USER _ -> DEL_USER_ STAT _ -> STAT_ OK -> OK_ @@ -651,7 +643,7 @@ instance ToJSON NotificationsMode where instance FromJSON NotificationsMode where parseJSON = strParseJSON "NotificationsMode" -instance ToField NotificationsMode where toField = toField . Binary . strEncode +instance ToField NotificationsMode where toField = toField . strEncode instance FromField NotificationsMode where fromField = blobFieldDecoder $ parseAll strP diff --git a/src/Simplex/Messaging/Agent/Stats.hs b/src/Simplex/Messaging/Agent/Stats.hs index 1d174622e..020c6a89c 100644 --- a/src/Simplex/Messaging/Agent/Stats.hs +++ b/src/Simplex/Messaging/Agent/Stats.hs @@ -1,4 +1,3 @@ -{-# LANGUAGE CPP #-} {-# LANGUAGE DuplicateRecordFields #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE NamedFieldPuns #-} @@ -12,17 +11,11 @@ import Data.Int (Int64) import Data.Map.Strict (Map) import qualified Data.Map.Strict as M import Simplex.Messaging.Agent.Protocol (UserId) +import Simplex.Messaging.Agent.Store.DB (FromField (..), ToField (..)) import Simplex.Messaging.Parsers (defaultJSON, fromTextField_) import Simplex.Messaging.Protocol (NtfServer, SMPServer, XFTPServer) import Simplex.Messaging.Util (decodeJSON, encodeJSON) import UnliftIO.STM -#if defined(dbPostgres) -import Database.PostgreSQL.Simple.FromField (FromField (..)) -import Database.PostgreSQL.Simple.ToField (ToField (..)) -#else -import Database.SQLite.Simple.FromField (FromField (..)) -import Database.SQLite.Simple.ToField (ToField (..)) -#endif data AgentSMPServerStats = AgentSMPServerStats { sentDirect :: TVar Int, -- successfully sent messages diff --git a/src/Simplex/Messaging/Agent/Store/AgentStore.hs b/src/Simplex/Messaging/Agent/Store/AgentStore.hs index 5dcda9c79..46a358745 100644 --- a/src/Simplex/Messaging/Agent/Store/AgentStore.hs +++ b/src/Simplex/Messaging/Agent/Store/AgentStore.hs @@ -262,7 +262,7 @@ import Simplex.Messaging.Agent.Stats import Simplex.Messaging.Agent.Store import Simplex.Messaging.Agent.Store.Common import qualified Simplex.Messaging.Agent.Store.DB as DB -import Simplex.Messaging.Agent.Store.DB (Binary (..), BoolInt (..)) +import Simplex.Messaging.Agent.Store.DB (Binary (..), BoolInt (..), FromField (..), ToField (..)) import qualified Simplex.Messaging.Crypto as C import Simplex.Messaging.Crypto.File (CryptoFile (..), CryptoFileArgs (..)) import Simplex.Messaging.Crypto.Ratchet (PQEncryption (..), PQSupport (..), RatchetX448, SkippedMsgDiff (..), SkippedMsgKeys) @@ -281,16 +281,12 @@ import qualified UnliftIO.Exception as E import UnliftIO.STM #if defined(dbPostgres) import Database.PostgreSQL.Simple (Only (..), Query, SqlError, (:.) (..)) -import Database.PostgreSQL.Simple.FromField (FromField (..)) import Database.PostgreSQL.Simple.Errors (constraintViolation) import Database.PostgreSQL.Simple.SqlQQ (sql) -import Database.PostgreSQL.Simple.ToField (ToField (..)) #else import Database.SQLite.Simple (FromRow (..), Only (..), Query (..), SQLError, ToRow (..), field, (:.) (..)) import qualified Database.SQLite.Simple as SQL -import Database.SQLite.Simple.FromField import Database.SQLite.Simple.QQ (sql) -import Database.SQLite.Simple.ToField (ToField (..)) #endif checkConstraint :: StoreError -> IO (Either StoreError a) -> IO (Either StoreError a) diff --git a/src/Simplex/Messaging/Agent/Store/DB.hs b/src/Simplex/Messaging/Agent/Store/DB.hs index f8c54e463..ade1f7e6d 100644 --- a/src/Simplex/Messaging/Agent/Store/DB.hs +++ b/src/Simplex/Messaging/Agent/Store/DB.hs @@ -3,11 +3,15 @@ module Simplex.Messaging.Agent.Store.DB #if defined(dbPostgres) ( module Simplex.Messaging.Agent.Store.Postgres.DB, + FromField (..), + ToField (..), ) where import Simplex.Messaging.Agent.Store.Postgres.DB #else ( module Simplex.Messaging.Agent.Store.SQLite.DB, + FromField (..), + ToField (..), ) where import Simplex.Messaging.Agent.Store.SQLite.DB diff --git a/src/Simplex/Messaging/Agent/Store/Postgres/DB.hs b/src/Simplex/Messaging/Agent/Store/Postgres/DB.hs index 9e597aef7..88b37f65a 100644 --- a/src/Simplex/Messaging/Agent/Store/Postgres/DB.hs +++ b/src/Simplex/Messaging/Agent/Store/Postgres/DB.hs @@ -4,6 +4,8 @@ module Simplex.Messaging.Agent.Store.Postgres.DB ( BoolInt (..), PSQL.Binary (..), PSQL.Connection, + FromField (..), + ToField (..), PSQL.connect, PSQL.close, execute, @@ -49,15 +51,15 @@ executeMany db q qs = void $ PSQL.executeMany db q qs -- used in FileSize instance FromField Word32 where fromField field dat = do - i <- fromField field dat - if i >= (0 :: Int64) + i :: Int64 <- fromField field dat + if i >= 0 && i <= fromIntegral (maxBound :: Word32) then pure (fromIntegral i :: Word32) else returnError ConversionFailed field "Negative value can't be converted to Word32" -- used in Version instance FromField Word16 where fromField field dat = do - i <- fromField field dat - if i >= (0 :: Int32) + i :: Int64 <- fromField field dat + if i >= 0 && i <= fromIntegral (maxBound :: Word16) then pure (fromIntegral i :: Word16) else returnError ConversionFailed field "Negative value can't be converted to Word16" diff --git a/src/Simplex/Messaging/Agent/Store/SQLite.hs b/src/Simplex/Messaging/Agent/Store/SQLite.hs index 585f40a0c..e472db488 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite.hs +++ b/src/Simplex/Messaging/Agent/Store/SQLite.hs @@ -69,32 +69,33 @@ data DBOpts = DBOpts { dbFilePath :: FilePath, dbKey :: ScrubbedBytes, keepKey :: Bool, - vacuum :: Bool + vacuum :: Bool, + track :: DB.TrackQueries } createDBStore :: DBOpts -> [Migration] -> MigrationConfirmation -> IO (Either MigrationError DBStore) -createDBStore DBOpts {dbFilePath, dbKey, keepKey, vacuum} migrations confirmMigrations = do +createDBStore DBOpts {dbFilePath, dbKey, keepKey, track, vacuum} migrations confirmMigrations = do let dbDir = takeDirectory dbFilePath createDirectoryIfMissing True dbDir - st <- connectSQLiteStore dbFilePath dbKey keepKey + st <- connectSQLiteStore dbFilePath dbKey keepKey track r <- migrateSchema st migrations confirmMigrations vacuum `onException` closeDBStore st case r of Right () -> pure $ Right st Left e -> closeDBStore st $> Left e -connectSQLiteStore :: FilePath -> ScrubbedBytes -> Bool -> IO DBStore -connectSQLiteStore dbFilePath key keepKey = do +connectSQLiteStore :: FilePath -> ScrubbedBytes -> Bool -> DB.TrackQueries -> IO DBStore +connectSQLiteStore dbFilePath key keepKey track = do dbNew <- not <$> doesFileExist dbFilePath - dbConn <- dbBusyLoop (connectDB dbFilePath key) + dbConn <- dbBusyLoop (connectDB dbFilePath key track) dbConnection <- newMVar dbConn dbKey <- newTVarIO $! storeKey key keepKey dbClosed <- newTVarIO False dbSem <- newTVarIO 0 pure DBStore {dbFilePath, dbKey, dbSem, dbConnection, dbNew, dbClosed} -connectDB :: FilePath -> ScrubbedBytes -> IO DB.Connection -connectDB path key = do - db <- DB.open path +connectDB :: FilePath -> ScrubbedBytes -> DB.TrackQueries -> IO DB.Connection +connectDB path key track = do + db <- DB.open path track prepare db `onException` DB.close db -- _printPragmas db path pure db @@ -127,12 +128,12 @@ openSQLiteStore_ DBStore {dbConnection, dbFilePath, dbKey, dbClosed} key keepKey bracketOnError (takeMVar dbConnection) (tryPutMVar dbConnection) - $ \DB.Connection {slow} -> do - DB.Connection {conn} <- connectDB dbFilePath key + $ \DB.Connection {slow, track} -> do + DB.Connection {conn} <- connectDB dbFilePath key track atomically $ do writeTVar dbClosed False writeTVar dbKey $! storeKey key keepKey - putMVar dbConnection DB.Connection {conn, slow} + putMVar dbConnection DB.Connection {conn, slow, track} reopenDBStore :: DBStore -> IO () reopenDBStore st@DBStore {dbKey, dbClosed} = diff --git a/src/Simplex/Messaging/Agent/Store/SQLite/DB.hs b/src/Simplex/Messaging/Agent/Store/SQLite/DB.hs index 7e8406d5c..59c282c46 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite/DB.hs +++ b/src/Simplex/Messaging/Agent/Store/SQLite/DB.hs @@ -11,6 +11,9 @@ module Simplex.Messaging.Agent.Store.SQLite.DB Binary (..), Connection (..), SlowQueryStats (..), + TrackQueries (..), + FromField (..), + ToField (..), open, close, execute, @@ -38,7 +41,7 @@ import Database.SQLite.Simple.ToField (ToField (..)) import Simplex.Messaging.Parsers (defaultJSON) import Simplex.Messaging.TMap (TMap) import qualified Simplex.Messaging.TMap as TM -import Simplex.Messaging.Util (diffToMilliseconds, tshow) +import Simplex.Messaging.Util (diffToMicroseconds, tshow) newtype BoolInt = BI {unBI :: Bool} deriving newtype (FromField, ToField) @@ -48,9 +51,13 @@ newtype Binary = Binary {fromBinary :: ByteString} data Connection = Connection { conn :: SQL.Connection, + track :: TrackQueries, slow :: TMap Query SlowQueryStats } +data TrackQueries = TQAll | TQSlow Int64 | TQOff + deriving (Eq) + data SlowQueryStats = SlowQueryStats { count :: Int64, timeMax :: Int64, @@ -59,22 +66,29 @@ data SlowQueryStats = SlowQueryStats } deriving (Show) -timeIt :: TMap Query SlowQueryStats -> Query -> IO a -> IO a -timeIt slow sql a = do - t <- getCurrentTime - r <- - a `catch` \e -> do - atomically $ TM.alter (Just . updateQueryErrors e) sql slow - throwIO e - t' <- getCurrentTime - let diff = diffToMilliseconds $ diffUTCTime t' t - when (diff > 1) $ atomically $ TM.alter (updateQueryStats diff) sql slow - pure r +timeIt :: Connection -> Query -> IO a -> IO a +timeIt Connection {slow, track} sql a + | track == TQOff = makeQuery + | otherwise = do + t <- getCurrentTime + r <- makeQuery + t' <- getCurrentTime + let diff = diffToMicroseconds $ diffUTCTime t' t + when (trackQuery diff) $ atomically $ TM.alter (updateQueryStats diff) sql slow + pure r where + makeQuery = + a `catch` \e -> do + atomically $ TM.alter (Just . updateQueryErrors e) sql slow + throwIO e + trackQuery diff = case track of + TQOff -> False + TQSlow t -> diff > t + TQAll -> True updateQueryErrors :: SomeException -> Maybe SlowQueryStats -> SlowQueryStats updateQueryErrors e Nothing = SlowQueryStats 0 0 0 $ M.singleton (tshow e) 1 - updateQueryErrors e (Just stats@SlowQueryStats {errs}) = - stats {errs = M.alter (Just . maybe 1 (+ 1)) (tshow e) errs} + updateQueryErrors e (Just st@SlowQueryStats {errs}) = + st {errs = M.alter (Just . maybe 1 (+ 1)) (tshow e) errs} updateQueryStats :: Int64 -> Maybe SlowQueryStats -> Maybe SlowQueryStats updateQueryStats diff Nothing = Just $ SlowQueryStats 1 diff diff M.empty updateQueryStats diff (Just SlowQueryStats {count, timeMax, timeAvg, errs}) = @@ -86,33 +100,33 @@ timeIt slow sql a = do errs } -open :: String -> IO Connection -open f = do +open :: String -> TrackQueries -> IO Connection +open f track = do conn <- SQL.open f slow <- TM.emptyIO - pure Connection {conn, slow} + pure Connection {conn, slow, track} close :: Connection -> IO () close = SQL.close . conn execute :: ToRow q => Connection -> Query -> q -> IO () -execute Connection {conn, slow} sql = timeIt slow sql . SQL.execute conn sql +execute c sql = timeIt c sql . SQL.execute (conn c) sql {-# INLINE execute #-} execute_ :: Connection -> Query -> IO () -execute_ Connection {conn, slow} sql = timeIt slow sql $ SQL.execute_ conn sql +execute_ c sql = timeIt c sql $ SQL.execute_ (conn c) sql {-# INLINE execute_ #-} executeMany :: ToRow q => Connection -> Query -> [q] -> IO () -executeMany Connection {conn, slow} sql = timeIt slow sql . SQL.executeMany conn sql +executeMany c sql = timeIt c sql . SQL.executeMany (conn c) sql {-# INLINE executeMany #-} query :: (ToRow q, FromRow r) => Connection -> Query -> q -> IO [r] -query Connection {conn, slow} sql = timeIt slow sql . SQL.query conn sql +query c sql = timeIt c sql . SQL.query (conn c) sql {-# INLINE query #-} query_ :: FromRow r => Connection -> Query -> IO [r] -query_ Connection {conn, slow} sql = timeIt slow sql $ SQL.query_ conn sql +query_ c sql = timeIt c sql $ SQL.query_ (conn c) sql {-# INLINE query_ #-} $(J.deriveJSON defaultJSON ''SlowQueryStats) diff --git a/src/Simplex/Messaging/Crypto.hs b/src/Simplex/Messaging/Crypto.hs index a955d0d8a..ef3548953 100644 --- a/src/Simplex/Messaging/Crypto.hs +++ b/src/Simplex/Messaging/Crypto.hs @@ -238,18 +238,11 @@ import Data.X509 import Data.X509.Validation (Fingerprint (..), getFingerprint) import GHC.TypeLits (ErrorMessage (..), KnownNat, Nat, TypeError, natVal, type (+)) import Network.Transport.Internal (decodeWord16, encodeWord16) -import Simplex.Messaging.Agent.Store.DB (Binary (..)) +import Simplex.Messaging.Agent.Store.DB (Binary (..), FromField (..), ToField (..)) import Simplex.Messaging.Encoding import Simplex.Messaging.Encoding.String import Simplex.Messaging.Parsers (blobFieldDecoder, parseAll, parseString) import Simplex.Messaging.Util ((<$?>)) -#if defined(dbPostgres) -import Database.PostgreSQL.Simple.FromField (FromField (..)) -import Database.PostgreSQL.Simple.ToField (ToField (..)) -#else -import Database.SQLite.Simple.FromField (FromField (..)) -import Database.SQLite.Simple.ToField (ToField (..)) -#endif -- | Cryptographic algorithms. data Algorithm = Ed25519 | Ed448 | X25519 | X448 diff --git a/src/Simplex/Messaging/Crypto/Ratchet.hs b/src/Simplex/Messaging/Crypto/Ratchet.hs index 310893de5..0ee4c75d0 100644 --- a/src/Simplex/Messaging/Crypto/Ratchet.hs +++ b/src/Simplex/Messaging/Crypto/Ratchet.hs @@ -111,7 +111,7 @@ import Data.Type.Equality import Data.Typeable (Typeable) import Data.Word (Word16, Word32) import Simplex.Messaging.Agent.QueryString -import Simplex.Messaging.Agent.Store.DB (Binary (..), BoolInt (..)) +import Simplex.Messaging.Agent.Store.DB (Binary (..), BoolInt (..), FromField (..), ToField (..)) import Simplex.Messaging.Crypto import Simplex.Messaging.Crypto.SNTRUP761.Bindings import Simplex.Messaging.Encoding @@ -121,13 +121,6 @@ import Simplex.Messaging.Util (($>>=), (<$?>)) import Simplex.Messaging.Version import Simplex.Messaging.Version.Internal import UnliftIO.STM -#if defined(dbPostgres) -import Database.PostgreSQL.Simple.FromField (FromField (..)) -import Database.PostgreSQL.Simple.ToField (ToField (..)) -#else -import Database.SQLite.Simple.FromField (FromField (..)) -import Database.SQLite.Simple.ToField (ToField (..)) -#endif -- e2e encryption headers version history: -- 1 - binary protocol encoding (1/1/2022) diff --git a/src/Simplex/Messaging/Crypto/SNTRUP761/Bindings.hs b/src/Simplex/Messaging/Crypto/SNTRUP761/Bindings.hs index 35e46e3de..82483491e 100644 --- a/src/Simplex/Messaging/Crypto/SNTRUP761/Bindings.hs +++ b/src/Simplex/Messaging/Crypto/SNTRUP761/Bindings.hs @@ -11,18 +11,12 @@ import Data.ByteArray (ScrubbedBytes) import qualified Data.ByteArray as BA import Data.ByteString (ByteString) import Foreign (nullPtr) +import Simplex.Messaging.Agent.Store.DB (FromField (..), ToField (..)) import Simplex.Messaging.Crypto.SNTRUP761.Bindings.Defines import Simplex.Messaging.Crypto.SNTRUP761.Bindings.FFI import Simplex.Messaging.Crypto.SNTRUP761.Bindings.RNG (withDRG) import Simplex.Messaging.Encoding import Simplex.Messaging.Encoding.String -#if defined(dbPostgres) -import Database.PostgreSQL.Simple.FromField -import Database.PostgreSQL.Simple.ToField -#else -import Database.SQLite.Simple.FromField -import Database.SQLite.Simple.ToField -#endif newtype KEMPublicKey = KEMPublicKey ByteString deriving (Eq, Show) diff --git a/src/Simplex/Messaging/Notifications/Protocol.hs b/src/Simplex/Messaging/Notifications/Protocol.hs index 96f8b337e..642465883 100644 --- a/src/Simplex/Messaging/Notifications/Protocol.hs +++ b/src/Simplex/Messaging/Notifications/Protocol.hs @@ -1,4 +1,3 @@ -{-# LANGUAGE CPP #-} {-# LANGUAGE DataKinds #-} {-# LANGUAGE GADTs #-} {-# LANGUAGE LambdaCase #-} @@ -29,6 +28,7 @@ import Data.Time.Clock.System import Data.Type.Equality import Data.Word (Word16) import Simplex.Messaging.Agent.Protocol (updateSMPServerHosts) +import Simplex.Messaging.Agent.Store.DB (FromField (..), ToField (..)) import qualified Simplex.Messaging.Crypto as C import Simplex.Messaging.Encoding import Simplex.Messaging.Encoding.String @@ -36,13 +36,6 @@ import Simplex.Messaging.Notifications.Transport (NTFVersion, ntfClientHandshake import Simplex.Messaging.Parsers (fromTextField_) import Simplex.Messaging.Protocol hiding (Command (..), CommandTag (..)) import Simplex.Messaging.Util (eitherToMaybe, (<$?>)) -#if defined(dbPostgres) -import Database.PostgreSQL.Simple.FromField (FromField (..)) -import Database.PostgreSQL.Simple.ToField (ToField (..)) -#else -import Database.SQLite.Simple.FromField (FromField (..)) -import Database.SQLite.Simple.ToField (ToField (..)) -#endif data NtfEntity = Token | Subscription deriving (Show) diff --git a/src/Simplex/Messaging/Notifications/Types.hs b/src/Simplex/Messaging/Notifications/Types.hs index dd6e99733..3daf97970 100644 --- a/src/Simplex/Messaging/Notifications/Types.hs +++ b/src/Simplex/Messaging/Notifications/Types.hs @@ -1,4 +1,3 @@ -{-# LANGUAGE CPP #-} {-# LANGUAGE DataKinds #-} {-# LANGUAGE DuplicateRecordFields #-} {-# LANGUAGE LambdaCase #-} @@ -11,19 +10,12 @@ import qualified Data.Attoparsec.ByteString.Char8 as A import Data.Text.Encoding (decodeLatin1, encodeUtf8) import Data.Time (UTCTime) import Simplex.Messaging.Agent.Protocol (ConnId, NotificationsMode (..), UserId) -import Simplex.Messaging.Agent.Store.DB (Binary (..)) +import Simplex.Messaging.Agent.Store.DB (Binary (..), FromField (..), ToField (..)) import qualified Simplex.Messaging.Crypto as C import Simplex.Messaging.Encoding import Simplex.Messaging.Notifications.Protocol import Simplex.Messaging.Parsers (blobFieldDecoder, fromTextField_) import Simplex.Messaging.Protocol (NotifierId, NtfServer, SMPServer) -#if defined(dbPostgres) -import Database.PostgreSQL.Simple.FromField (FromField (..)) -import Database.PostgreSQL.Simple.ToField (ToField (..)) -#else -import Database.SQLite.Simple.FromField (FromField (..)) -import Database.SQLite.Simple.ToField (ToField (..)) -#endif data NtfTknAction = NTARegister diff --git a/src/Simplex/Messaging/Server.hs b/src/Simplex/Messaging/Server.hs index 00009ec69..ddaeb789e 100644 --- a/src/Simplex/Messaging/Server.hs +++ b/src/Simplex/Messaging/Server.hs @@ -1150,9 +1150,10 @@ client ms clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId, procThreads} = do labelMyThread . B.unpack $ "client $" <> encode sessionId <> " commands" + let THandleParams {thVersion} = thParams' forever $ atomically (readTBQueue rcvQ) - >>= mapM processCommand + >>= mapM (processCommand thVersion) >>= mapM_ reply . L.nonEmpty . catMaybes . L.toList where reply :: MonadIO m => NonEmpty (Transmission BrokerMsg) -> m () @@ -1237,8 +1238,8 @@ client mkIncProxyStats ps psOwn own sel = do incStat $ sel ps when own $ incStat $ sel psOwn - processCommand :: (Maybe (StoreQueue s, QueueRec), Transmission Cmd) -> M (Maybe (Transmission BrokerMsg)) - processCommand (q_, (corrId, entId, cmd)) = case cmd of + processCommand :: VersionSMP -> (Maybe (StoreQueue s, QueueRec), Transmission Cmd) -> M (Maybe (Transmission BrokerMsg)) + processCommand clntVersion (q_, (corrId, entId, cmd)) = case cmd of Cmd SProxiedClient command -> processProxiedCmd (corrId, entId, command) Cmd SSender command -> Just <$> case command of SKEY sKey -> @@ -1499,7 +1500,7 @@ client sendMessage :: MsgFlags -> MsgBody -> StoreQueue s -> QueueRec -> M (Transmission BrokerMsg) sendMessage msgFlags msgBody q qr - | B.length msgBody > maxMessageLength thVersion = do + | B.length msgBody > maxMessageLength clntVersion = do stats <- asks serverStats incStat $ msgSentLarge stats pure $ err LARGE_MSG @@ -1538,7 +1539,6 @@ client liftIO $ updatePeriodStats (activeQueues stats) (recipientId' q) pure ok where - THandleParams {thVersion} = thParams' mkMessage :: MsgId -> C.MaxLenBS MaxMessageLen -> IO Message mkMessage msgId body = do msgTs <- getSystemTime @@ -1647,7 +1647,7 @@ client Left r -> pure r -- rejectOrVerify filters allowed commands, no need to repeat it here. -- INTERNAL is used because processCommand never returns Nothing for sender commands (could be extracted for better types). - Right t''@(_, (corrId', entId', _)) -> fromMaybe (corrId', entId', ERR INTERNAL) <$> lift (processCommand t'') + Right t''@(_, (corrId', entId', _)) -> fromMaybe (corrId', entId', ERR INTERNAL) <$> lift (processCommand fwdVersion t'') -- encode response r' <- case batchTransmissions (batch clntTHParams) (blockSize clntTHParams) [Right (Nothing, encodeTransmission clntTHParams r)] of [] -> throwE INTERNAL -- at least 1 item is guaranteed from NonEmpty/Right diff --git a/src/Simplex/RemoteControl/Types.hs b/src/Simplex/RemoteControl/Types.hs index 666878c30..bc191824a 100644 --- a/src/Simplex/RemoteControl/Types.hs +++ b/src/Simplex/RemoteControl/Types.hs @@ -1,4 +1,3 @@ -{-# LANGUAGE CPP #-} {-# LANGUAGE DataKinds #-} {-# LANGUAGE DeriveAnyClass #-} {-# LANGUAGE DuplicateRecordFields #-} diff --git a/tests/AgentTests/FunctionalAPITests.hs b/tests/AgentTests/FunctionalAPITests.hs index 1a6ff5173..5510307a3 100644 --- a/tests/AgentTests/FunctionalAPITests.hs +++ b/tests/AgentTests/FunctionalAPITests.hs @@ -2055,8 +2055,8 @@ testAsyncCommands sqSecured alice bob baseId = ackMessageAsync alice "7" bobId (baseId + 4) Nothing get alice =##> \case ("7", _, OK) -> True; _ -> False deleteConnectionAsync alice False bobId - get alice =##> \case ("", c, DEL_RCVQ _ _ Nothing) -> c == bobId; _ -> False - get alice =##> \case ("", c, DEL_CONN) -> c == bobId; _ -> False + get alice =##> \case ("", "", DEL_RCVQS [(c, _, _, Nothing)]) -> c == bobId; _ -> False + get alice =##> \case ("", "", DEL_CONNS [c]) -> c == bobId; _ -> False liftIO $ noMessages alice "nothing else should be delivered to alice" where msgId = subtract baseId @@ -2123,12 +2123,9 @@ testDeleteConnectionAsync t = runRight_ $ do deleteConnectionsAsync a False connIds nGet a =##> \case ("", "", DOWN {}) -> True; _ -> False - get a =##> \case ("", c, DEL_RCVQ _ _ (Just (BROKER _ e))) -> c `elem` connIds && (e == TIMEOUT || e == NETWORK); _ -> False - get a =##> \case ("", c, DEL_RCVQ _ _ (Just (BROKER _ e))) -> c `elem` connIds && (e == TIMEOUT || e == NETWORK); _ -> False - get a =##> \case ("", c, DEL_RCVQ _ _ (Just (BROKER _ e))) -> c `elem` connIds && (e == TIMEOUT || e == NETWORK); _ -> False - get a =##> \case ("", c, DEL_CONN) -> c `elem` connIds; _ -> False - get a =##> \case ("", c, DEL_CONN) -> c `elem` connIds; _ -> False - get a =##> \case ("", c, DEL_CONN) -> c `elem` connIds; _ -> False + let delOk = \case (c, _, _, Just (BROKER _ e)) -> c `elem` connIds && (e == TIMEOUT || e == NETWORK); _ -> False + get a =##> \case ("", "", DEL_RCVQS rs) -> length rs == 3 && all delOk rs; _ -> False + get a =##> \case ("", "", DEL_CONNS cs) -> length cs == 3 && all (`elem` connIds) cs; _ -> False liftIO $ noMessages a "nothing else should be delivered to alice" testWaitDeliveryNoPending :: ATransport -> IO () @@ -2147,8 +2144,8 @@ testWaitDeliveryNoPending t = withAgentClients2 $ \alice bob -> ackMessage alice bobId (baseId + 2) Nothing deleteConnectionsAsync alice True [bobId] - get alice =##> \case ("", cId, DEL_RCVQ _ _ Nothing) -> cId == bobId; _ -> False - get alice =##> \case ("", cId, DEL_CONN) -> cId == bobId; _ -> False + get alice =##> \case ("", "", DEL_RCVQS [(cId, _, _, Nothing)]) -> cId == bobId; _ -> False + get alice =##> \case ("", "", DEL_CONNS [cId]) -> cId == bobId; _ -> False 3 <- msgId <$> sendMessage bob aliceId SMP.noMsgFlags "message 2" get bob =##> \case ("", cId, MERR mId (SMP _ AUTH)) -> cId == aliceId && mId == (baseId + 3); _ -> False @@ -2184,14 +2181,14 @@ testWaitDelivery t = 3 <- msgId <$> sendMessage alice bobId SMP.noMsgFlags "how are you?" 4 <- msgId <$> sendMessage alice bobId SMP.noMsgFlags "message 1" deleteConnectionsAsync alice True [bobId] - get alice =##> \case ("", cId, DEL_RCVQ _ _ (Just (BROKER _ e))) -> cId == bobId && (e == TIMEOUT || e == NETWORK); _ -> False + get alice =##> \case ("", "", DEL_RCVQS [(cId, _, _, Just (BROKER _ e))]) -> cId == bobId && (e == TIMEOUT || e == NETWORK); _ -> False liftIO $ noMessages alice "nothing else should be delivered to alice" liftIO $ noMessages bob "nothing else should be delivered to bob" withSmpServerStoreLogOn t testPort $ \_ -> runRight_ $ do get alice ##> ("", bobId, SENT $ baseId + 3) get alice ##> ("", bobId, SENT $ baseId + 4) - get alice =##> \case ("", cId, DEL_CONN) -> cId == bobId; _ -> False + get alice =##> \case ("", "", DEL_CONNS [cId]) -> cId == bobId; _ -> False liftIO $ getInAnyOrder @@ -2231,8 +2228,8 @@ testWaitDeliveryAUTHErr t = ackMessage alice bobId (baseId + 2) Nothing deleteConnectionsAsync bob False [aliceId] - get bob =##> \case ("", cId, DEL_RCVQ _ _ Nothing) -> cId == aliceId; _ -> False - get bob =##> \case ("", cId, DEL_CONN) -> cId == aliceId; _ -> False + get bob =##> \case ("", "", DEL_RCVQS [(cId, _, _, Nothing)]) -> cId == aliceId; _ -> False + get bob =##> \case ("", "", DEL_CONNS [cId]) -> cId == aliceId; _ -> False pure (aliceId, bobId) @@ -2241,14 +2238,14 @@ testWaitDeliveryAUTHErr t = 3 <- msgId <$> sendMessage alice bobId SMP.noMsgFlags "how are you?" 4 <- msgId <$> sendMessage alice bobId SMP.noMsgFlags "message 1" deleteConnectionsAsync alice True [bobId] - get alice =##> \case ("", cId, DEL_RCVQ _ _ (Just (BROKER _ e))) -> cId == bobId && (e == TIMEOUT || e == NETWORK); _ -> False + get alice =##> \case ("", "", DEL_RCVQS [(cId, _, _, Just (BROKER _ e))]) -> cId == bobId && (e == TIMEOUT || e == NETWORK); _ -> False liftIO $ noMessages alice "nothing else should be delivered to alice" liftIO $ noMessages bob "nothing else should be delivered to bob" withSmpServerStoreLogOn t testPort $ \_ -> do get alice =##> \case ("", cId, MERR mId (SMP _ AUTH)) -> cId == bobId && mId == (baseId + 3); _ -> False get alice =##> \case ("", cId, MERR mId (SMP _ AUTH)) -> cId == bobId && mId == (baseId + 4); _ -> False - get alice =##> \case ("", cId, DEL_CONN) -> cId == bobId; _ -> False + get alice =##> \case ("", "", DEL_CONNS [cId]) -> cId == bobId; _ -> False liftIO $ noMessages alice "nothing else should be delivered to alice" liftIO $ noMessages bob "nothing else should be delivered to bob" @@ -2281,8 +2278,8 @@ testWaitDeliveryTimeout t = 3 <- msgId <$> sendMessage alice bobId SMP.noMsgFlags "how are you?" 4 <- msgId <$> sendMessage alice bobId SMP.noMsgFlags "message 1" deleteConnectionsAsync alice True [bobId] - get alice =##> \case ("", cId, DEL_RCVQ _ _ (Just (BROKER _ e))) -> cId == bobId && (e == TIMEOUT || e == NETWORK); _ -> False - get alice =##> \case ("", cId, DEL_CONN) -> cId == bobId; _ -> False + get alice =##> \case ("", "", DEL_RCVQS [(cId, _, _, Just (BROKER _ e))]) -> cId == bobId && (e == TIMEOUT || e == NETWORK); _ -> False + get alice =##> \case ("", "", DEL_CONNS [cId]) -> cId == bobId; _ -> False liftIO $ noMessages alice "nothing else should be delivered to alice" liftIO $ noMessages bob "nothing else should be delivered to bob" @@ -2321,8 +2318,8 @@ testWaitDeliveryTimeout2 t = 3 <- msgId <$> sendMessage alice bobId SMP.noMsgFlags "how are you?" 4 <- msgId <$> sendMessage alice bobId SMP.noMsgFlags "message 1" deleteConnectionsAsync alice True [bobId] - get alice =##> \case ("", cId, DEL_RCVQ _ _ (Just (BROKER _ e))) -> cId == bobId && (e == TIMEOUT || e == NETWORK); _ -> False - get alice =##> \case ("", cId, DEL_CONN) -> cId == bobId; _ -> False + get alice =##> \case ("", "", DEL_RCVQS [(cId, _, _, Just (BROKER _ e))]) -> cId == bobId && (e == TIMEOUT || e == NETWORK); _ -> False + get alice =##> \case ("", "", DEL_CONNS [cId]) -> cId == bobId; _ -> False liftIO $ noMessages alice "nothing else should be delivered to alice" liftIO $ noMessages bob "nothing else should be delivered to bob" @@ -2430,8 +2427,8 @@ testUsers = (aId', bId') <- makeConnectionForUsers a auId b 1 exchangeGreetings a bId' b aId' deleteUser a auId True - get a =##> \case ("", c, DEL_RCVQ _ _ Nothing) -> c == bId'; _ -> False - get a =##> \case ("", c, DEL_CONN) -> c == bId'; _ -> False + get a =##> \case ("", "", DEL_RCVQS [(c, _, _, Nothing)]) -> c == bId'; _ -> False + get a =##> \case ("", "", DEL_CONNS [c]) -> c == bId'; _ -> False nGet a =##> \case ("", "", DEL_USER u) -> u == auId; _ -> False exchangeGreetingsMsgId 4 a bId b aId liftIO $ noMessages a "nothing else should be delivered to alice" @@ -2462,8 +2459,8 @@ testUsersNoServer t = withAgentClientsCfg2 aCfg agentCfg $ \a b -> do nGet b =##> \case ("", "", DOWN _ cs) -> length cs == 2; _ -> False runRight_ $ do deleteUser a auId True - get a =##> \case ("", c, DEL_RCVQ _ _ (Just (BROKER _ e))) -> c == bId' && (e == TIMEOUT || e == NETWORK); _ -> False - get a =##> \case ("", c, DEL_CONN) -> c == bId'; _ -> False + get a =##> \case ("", "", DEL_RCVQS [(c, _, _, Just (BROKER _ e))]) -> c == bId' && (e == TIMEOUT || e == NETWORK); _ -> False + get a =##> \case ("", "", DEL_CONNS [c]) -> c == bId'; _ -> False nGet a =##> \case ("", "", DEL_USER u) -> u == auId; _ -> False liftIO $ noMessages a "nothing else should be delivered to alice" withSmpServerStoreLogOn t testPort $ \_ -> runRight_ $ do @@ -2581,9 +2578,8 @@ testSwitchDelete servers = liftIO $ rcvSwchStatuses' stats `shouldMatchList` [Just RSSwitchStarted] phaseRcv a bId SPStarted [Just RSSendingQADD, Nothing] deleteConnectionAsync a False bId - get a =##> \case ("", c, DEL_RCVQ _ _ Nothing) -> c == bId; _ -> False - get a =##> \case ("", c, DEL_RCVQ _ _ Nothing) -> c == bId; _ -> False - get a =##> \case ("", c, DEL_CONN) -> c == bId; _ -> False + get a =##> \case ("", "", DEL_RCVQS [(c, _, _, Nothing), (c', _, _, Nothing)]) -> c == bId && c' == bId; _ -> False + get a =##> \case ("", "", DEL_CONNS [c]) -> c == bId; _ -> False liftIO $ noMessages a "nothing else should be delivered to alice" testAbortSwitchStarted :: HasCallStack => InitialAgentServers -> IO () @@ -3104,7 +3100,7 @@ insertUser :: DBStore -> IO () insertUser st = withTransaction st (`DB.execute_` "INSERT INTO users DEFAULT VALUES") #else createStore :: String -> IO (Either MigrationError DBStore) -createStore dbPath = createAgentStore (DBOpts dbPath "" False True) MCError +createStore dbPath = createAgentStore (DBOpts dbPath "" False True DB.TQOff) MCError insertUser :: DBStore -> IO () insertUser st = withTransaction st (`DB.execute_` "INSERT INTO users (user_id) VALUES (1)") diff --git a/tests/AgentTests/MigrationTests.hs b/tests/AgentTests/MigrationTests.hs index 5ad4f101d..1a879eca7 100644 --- a/tests/AgentTests/MigrationTests.hs +++ b/tests/AgentTests/MigrationTests.hs @@ -228,7 +228,8 @@ createStore randSuffix migrations confirmMigrations = do dbFilePath = testDB randSuffix, dbKey = "", keepKey = False, - vacuum = True + vacuum = True, + track = DB.TQOff } createDBStore dbOpts migrations confirmMigrations diff --git a/tests/AgentTests/SQLiteTests.hs b/tests/AgentTests/SQLiteTests.hs index 84f30ff96..6950f3379 100644 --- a/tests/AgentTests/SQLiteTests.hs +++ b/tests/AgentTests/SQLiteTests.hs @@ -70,7 +70,7 @@ withStore2 = before connect2 . after (removeStore . fst) connect2 :: IO (DBStore, DBStore) connect2 = do s1@DBStore {dbFilePath} <- createStore' - s2 <- connectSQLiteStore dbFilePath "" False + s2 <- connectSQLiteStore dbFilePath "" False DB.TQOff pure (s1, s2) createStore' :: IO DBStore @@ -81,7 +81,7 @@ createEncryptedStore key keepKey = do -- Randomize DB file name to avoid SQLite IO errors supposedly caused by asynchronous -- IO operations on multiple similarly named files; error seems to be environment specific r <- randomIO :: IO Word32 - Right st <- createDBStore (DBOpts (testDB <> show r) key keepKey True) Migrations.app MCError + Right st <- createDBStore (DBOpts (testDB <> show r) key keepKey True DB.TQOff) Migrations.app MCError withTransaction' st (`SQL.execute_` "INSERT INTO users (user_id) VALUES (1);") pure st diff --git a/tests/AgentTests/SchemaDump.hs b/tests/AgentTests/SchemaDump.hs index 75e89d00e..b2ddbdbce 100644 --- a/tests/AgentTests/SchemaDump.hs +++ b/tests/AgentTests/SchemaDump.hs @@ -12,6 +12,7 @@ import Database.SQLite.Simple (Only (..)) import qualified Database.SQLite.Simple as SQL import Simplex.Messaging.Agent.Store.SQLite import Simplex.Messaging.Agent.Store.SQLite.Common (withTransaction') +import Simplex.Messaging.Agent.Store.SQLite.DB (TrackQueries (..)) import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations import Simplex.Messaging.Agent.Store.Shared (Migration (..), MigrationConfirmation (..), MigrationsToRun (..), toDownMigration) import Simplex.Messaging.Util (ifM) @@ -49,7 +50,7 @@ testVerifySchemaDump :: IO () testVerifySchemaDump = do savedSchema <- ifM (doesFileExist appSchema) (readFile appSchema) (pure "") savedSchema `deepseq` pure () - void $ createDBStore (DBOpts testDB "" False True) Migrations.app MCConsole + void $ createDBStore (DBOpts testDB "" False True TQOff) Migrations.app MCConsole getSchema testDB appSchema `shouldReturn` savedSchema removeFile testDB @@ -57,7 +58,7 @@ testVerifyLintFKeyIndexes :: IO () testVerifyLintFKeyIndexes = do savedLint <- ifM (doesFileExist appLint) (readFile appLint) (pure "") savedLint `deepseq` pure () - void $ createDBStore (DBOpts testDB "" False True) Migrations.app MCConsole + void $ createDBStore (DBOpts testDB "" False True TQOff) Migrations.app MCConsole getLintFKeyIndexes testDB "tests/tmp/agent_lint.sql" `shouldReturn` savedLint removeFile testDB @@ -70,7 +71,7 @@ withTmpFiles = testSchemaMigrations :: IO () testSchemaMigrations = do let noDownMigrations = dropWhileEnd (\Migration {down} -> isJust down) Migrations.app - Right st <- createDBStore (DBOpts testDB "" False True) noDownMigrations MCError + Right st <- createDBStore (DBOpts testDB "" False True TQOff) noDownMigrations MCError mapM_ (testDownMigration st) $ drop (length noDownMigrations) Migrations.app closeDBStore st removeFile testDB @@ -93,7 +94,7 @@ testSchemaMigrations = do testUsersMigrationNew :: IO () testUsersMigrationNew = do - Right st <- createDBStore (DBOpts testDB "" False True) Migrations.app MCError + Right st <- createDBStore (DBOpts testDB "" False True TQOff) Migrations.app MCError withTransaction' st (`SQL.query_` "SELECT user_id FROM users;") `shouldReturn` ([] :: [Only Int]) closeDBStore st @@ -101,11 +102,11 @@ testUsersMigrationNew = do testUsersMigrationOld :: IO () testUsersMigrationOld = do let beforeUsers = takeWhile (("m20230110_users" /=) . name) Migrations.app - Right st <- createDBStore (DBOpts testDB "" False True) beforeUsers MCError + Right st <- createDBStore (DBOpts testDB "" False True TQOff) beforeUsers MCError withTransaction' st (`SQL.query_` "SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'users';") `shouldReturn` ([] :: [Only String]) closeDBStore st - Right st' <- createDBStore (DBOpts testDB "" False True) Migrations.app MCYesUp + Right st' <- createDBStore (DBOpts testDB "" False True TQOff) Migrations.app MCYesUp withTransaction' st' (`SQL.query_` "SELECT user_id FROM users;") `shouldReturn` ([Only (1 :: Int)]) closeDBStore st'