diff --git a/.github/dependabot.yml b/.github/dependabot.yml deleted file mode 100644 index 8617b871e4..0000000000 --- a/.github/dependabot.yml +++ /dev/null @@ -1,96 +0,0 @@ -# Configure general package update PRs filed by dependabot. -# Security update PRs are generally unaffected by these settings. -version: 2 - -updates: - - # "pip" is the correct setting for poetry, per - # https://docs.github.com/en/code-security/dependabot/dependabot-version-updates/configuration-options-for-the-dependabot.yml-file#package-ecosystem - package-ecosystem: "pip" - directory: "/" - # Only bump the minimum and maximum versions of dependencies in our - # lockfile. The maximum version in `pyproject.toml` may be bumped, but the - # minimum may not be. - # - # We want to avoid bumping the minimum version in `pyproject.toml` unless - # necessary, as it requires our supported Linux distributions to have - # those versions in their repos first. - versioning-strategy: "increase-if-necessary" - # Group together all updates for this ecosystem into a single, big PR. - # - # Doing this prevents us from needing to review a PR per dependency upgrade - # (which can become burdensome). - # - # Splitting up each ecosystem into their own PR (rather than putting ALL - # upgrades into a single PR) allows us to more quicker determine the culprit - # when CI fails. - groups: - all-python-dependencies: - applies-to: version-updates - patterns: - # Update all non-transitive dependencies. - - "*" - open-pull-requests-limit: 1 - # Prevent pulling packages that were recently updated to help mitigate - # supply chain attacks. 14 days was taken from the recommendation at - # https://blog.yossarian.net/2025/11/21/We-should-all-be-using-dependency-cooldowns - # where the author noted that 9/10 attacks would have been mitigated by a - # two week cooldown. - # - # The cooldown only applies to general updates; security updates will still - # be pulled in as soon as possible. - cooldown: - default-days: 14 - schedule: - interval: "monthly" - time: "08:00" - timezone: "Europe/London" - - - package-ecosystem: "cargo" - directory: "/" - open-pull-requests-limit: 1 - groups: - all-rust-dependencies: - applies-to: version-updates - patterns: - - "*" - schedule: - interval: "monthly" - time: "08:00" - timezone: "Europe/London" - cooldown: - default-days: 14 - - - package-ecosystem: "github-actions" - directory: "/" - open-pull-requests-limit: 1 - groups: - all-github-actions: - applies-to: version-updates - patterns: - - "*" - schedule: - interval: "monthly" - time: "08:00" - timezone: "Europe/London" - cooldown: - default-days: 14 - - - package-ecosystem: "gomod" - directory: "/complement" - open-pull-requests-limit: 1 - groups: - all-go-dependencies: - applies-to: version-updates - patterns: - - "*" - schedule: - interval: "monthly" - time: "08:00" - timezone: "Europe/London" - cooldown: - default-days: 14 - - # Note: "docker" is purposefully excluded from this list. - # - # Our docker containers should be kept up-to-date using container scanning - # technology instead of dependabot. \ No newline at end of file diff --git a/.github/workflows/beeper-ci.yaml b/.github/workflows/beeper-ci.yaml new file mode 100644 index 0000000000..207e34f146 --- /dev/null +++ b/.github/workflows/beeper-ci.yaml @@ -0,0 +1,143 @@ +name: Beep + +on: + push: + branches: ["beeper", "beeper-*"] + pull_request: + + +jobs: + lint-style: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v7 + - uses: actions/setup-python@v7 + with: + python-version: "3.13" + - run: pip install poetry + - run: poetry install + - run: poetry run ruff check --output-format=github . + + lint-types: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v7 + - uses: actions/setup-python@v7 + with: + python-version: "3.13" + - run: pip install poetry + - run: poetry install --extras all + - run: poetry run mypy + + # Tests + + test-trial: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v7 + - uses: actions/setup-python@v7 + with: + python-version: "3.13" + - run: pip install poetry + - run: poetry install --extras all + - run: poetry run trial -j4 tests + + test-sytest: + runs-on: ubuntu-latest + container: + image: matrixdotorg/sytest-synapse:bookworm + volumes: + - ${{ github.workspace }}:/src + env: + SYTEST_BRANCH: 747315856d24eee846c6923b696fff9fae55ca0f + TOP: ${{ github.workspace }} + POSTGRES: 1 + MULTI_POSTGRES: 1 + WOKRERS: 1 + steps: + - uses: actions/checkout@v7 + - name: Run SyTest + run: /bootstrap.sh synapse + working-directory: /src + - name: Summarise results.tap + if: ${{ always() }} + run: /sytest/scripts/tap_to_gha.pl /logs/results.tap + - name: Upload SyTest logs + uses: actions/upload-artifact@v7 + if: ${{ always() }} + with: + name: Sytest Logs - ${{ job.status }} - (${{ join(matrix.*, ', ') }}) + path: | + /logs/results.tap + /logs/**/*.log* + + test-complement: + runs-on: ubuntu-latest + if: false + steps: + - uses: actions/checkout@v7 + - uses: actions/checkout@v7 + with: + repository: matrix-org/complement + path: complement + ref: 0116400a24dfaeb5e7eb9d19f270c94b7d134122 + - name: Install complement dependencies + run: |- + sudo apt-get -qq update + sudo apt-get install -qqy libolm3 libolm-dev + go install -v github.com/gotesttools/gotestfmt/v2/cmd/gotestfmt@latest + pip install poetry + poetry install + - name: Run Complement + run: ./scripts-dev/complement.sh + env: + COMPLEMENT_DIR: complement + + # Builds + + build-python: + runs-on: ubuntu-latest + env: + DOCKER_BUILDKIT: 1 + steps: + - uses: actions/checkout@v7 + - uses: docker/setup-qemu-action@v4 + - uses: docker/setup-buildx-action@v4 + - uses: docker/login-action@v4 + with: + registry: ${{ secrets.CI_REGISTRY }} + username: ${{ secrets.CI_REGISTRY_USER }} + password: ${{ secrets.CI_REGISTRY_PASSWORD }} + - run: |- + if [ "${{ github.ref_name }}" = "beeper" ]; then + tag=$(cat pyproject.toml | grep -E "^version =" | sed -E 's/^version = "(.+)"$/\1/') + else + tag="${{ github.head_ref || github.ref_name }}" + tag="${tag//\//-}" + fi + + docker buildx build \ + --push \ + --platform linux/amd64 \ + --tag ${{ secrets.CI_REGISTRY }}/synapse:$tag-${{ github.sha }} \ + -f docker/Dockerfile \ + . + + if [ "${{ github.ref_name }}" = "beeper" ]; then + docker pull ${{ secrets.CI_REGISTRY }}/synapse:$tag-${{ github.sha }} + docker tag \ + ${{ secrets.CI_REGISTRY }}/synapse:$tag-${{ github.sha }} \ + ${{ secrets.CI_REGISTRY }}/synapse:latest + docker push ${{ secrets.CI_REGISTRY }}/synapse:latest + fi + + # Ensure the image works properly + docker run \ + --entrypoint '' \ + ${{ secrets.CI_REGISTRY }}/synapse:$tag-${{ github.sha }} \ + python -m synapse.app.homeserver --help + + echo "Pushed image: synapse:$tag-${{ github.sha }}" + if [ "${{ github.ref_name }}" = "beeper" ]; then + echo "Pushed image: synapse:latest" + fi diff --git a/.github/workflows/schema_diff.yml b/.github/workflows/schema_diff.yml deleted file mode 100644 index e511ba64fc..0000000000 --- a/.github/workflows/schema_diff.yml +++ /dev/null @@ -1,128 +0,0 @@ -name: Schema Diff # If this is changed, need to update `schema_diff_comment.yml` as well - -on: - pull_request: - paths: - - synapse/storage/schema/*/delta/** - - synapse/storage/schema/*/full_schemas/** - - .github/workflows/schema_diff.yml - -concurrency: - group: ${{ github.workflow }}-${{ github.ref }} - cancel-in-progress: true - -permissions: - contents: read - -jobs: - # Posts a GitHub PR comment that shows what the effective change to the schema is. - # Provides an excuse to run the `make_full_schema.sh` script in CI (so we keep it working) - # and can act as a review aid for schema changes, letting you easily see the diff of the - # end result, even when background updates or complex schema deltas are present. - show-schema-diff: - name: Show schema diff - runs-on: ubuntu-latest - - permissions: - pull-requests: write - - steps: - - name: Start postgres with a faked clock - background: true - id: postgres - # Use faketime here for schema deltas that are wall-clock sensitive under Postgres - # For SQLite, faketime is used when invoking `make_full_schema.sh` within the script - run: | - # Build a docker image with faketime - mkdir /tmp/postgres-faketime - cat > /tmp/postgres-faketime/Dockerfile <<'EOF' - FROM postgres:14-alpine - RUN apk add --no-cache libfaketime - - # It seems like it could be harmful to fake the monotonic timer - # as it might prevent deadlock detection, etc. - # But not sure, just doing out of precaution. - ENV FAKETIME_DONT_FAKE_MONOTONIC=1 - ENTRYPOINT ["faketime", "-f", "2001-05-25 12:42:42", "docker-entrypoint.sh"] - CMD ["postgres"] - EOF - docker build -t localhost/postgres-faketime /tmp/postgres-faketime - - # Run it in the background - docker run -d --name postgres -p 5432:5432 \ - -e POSTGRES_PASSWORD=postgres \ - -e POSTGRES_INITDB_ARGS="--lc-collate C --lc-ctype C --encoding UTF8" \ - --health-cmd pg_isready --health-interval 10s \ - --health-timeout 5s --health-retries 5 \ - localhost/postgres-faketime - - - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 - with: - fetch-depth: 0 - - - name: Install PostgreSQL client and faketime - run: sudo apt-get -qq install postgresql-client faketime - - - uses: matrix-org/setup-python-poetry@5bbf6603c5c930615ec8a29f1b5d7d258d905aa4 # v2.0.0 - with: - poetry-version: "2.4.1" - extras: "postgres" - python-version: "3.x" - - - name: Wait for Postgres to be up - run: | - until [ "$(docker inspect -f '{{.State.Health.Status}}' postgres)" = healthy ]; do sleep 2; done - - - name: Generate schema diff - id: schema_diff - env: - PGHOST: localhost - PGUSER: postgres - PGPASSWORD: postgres - run: | - poetry run python .ci/scripts/schema_diff.py \ - --base origin/develop \ - > "${{ runner.temp }}/schema_diff.md" - - - name: Stop postgres - cancel: postgres - - # If the generation step failed, write an error message so the sticky - # comment step still has a file to read. - - name: Ensure output file exists on failure - if: always() && steps.schema_diff.outcome == 'failure' - run: | - echo "⚠️ Schema diff generation failed. See job logs for details." \ - > "${{ runner.temp }}/schema_diff.md" - - # Post a comment. - # - # For same-repo PRs, we can do this directly and we are done. - - name: Post sticky PR comment - uses: marocchino/sticky-pull-request-comment@28ad303d6acb4ccfc50f20c0044188e7e294b8c4 # v3 - # Only run for same-repo PRs - if: always() && github.event.pull_request.head.repo.full_name == github.repository - with: - header: schema-diff - path: ${{ runner.temp }}/schema_diff.md - - # For PRs from forks, we only have a read-only `GITHUB_TOKEN` here and need to work around - # by triggering a `workflow_run` onto the main repo. - # - # We upload an artifact and the `schema_diff_comment.yml` workflow will be triggered - # on the main repo by this workflow exiting. - - name: Save PR number for forked PRs - # Only run for fork PRs - if: always() && github.event.pull_request.head.repo.full_name != github.repository - run: echo "${{ github.event.pull_request.number }}" > "${{ runner.temp }}/pr_number" - - - name: Upload schema diff artifact for forked PRs - # Only run for fork PRs - if: always() && github.event.pull_request.head.repo.full_name != github.repository - uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 - with: - name: schema-diff - path: | - ${{ runner.temp }}/schema_diff.md - ${{ runner.temp }}/pr_number - retention-days: 1 diff --git a/README.md b/README.md new file mode 100644 index 0000000000..1867cbc485 --- /dev/null +++ b/README.md @@ -0,0 +1,67 @@ +# Synapse: Beeper Edition + +This is Beeper's custom version of synapse, we rebase roughly 25 commits on top of each upstream release with a few Beeper specific modifications. We also have an actual Synapse fork here: [**beeper/synapse-fork**](https://github.com/beeper/synapse-fork) which is where we make changes we expect to merge into upstream. + +## Branching strategy + +We have a bunch of branches. The main development branch for Beeper is +the `beeper` branch. Push commits here and they get built for our +Docker registry. The GitHub Actions logs include the specific image +tag, which is based on the original upstream version as well as the +commit hash. + +We also have versioned `beeper-x.y.z` branches. These are used to +archive the history of the `beeper` branch as it is rebased onto newer +upstream Synapse versions. When rebasing, we copy the current `beeper` +branch back onto the old `beeper-x.y.z` branch for historical +reference, then create a new `beeper-x.y.z` branch based on the new +upstream version and rebase the `beeper` branch onto it, then +force-push that resulting commit into the original `beeper` branch, +which will become the new development target. + +We also have `upstream-x.y.z` branches that just track the upstream +tags that we use as bases for Beeper changes, see the rebase flow +below. + +## CI setup + +Note that we have a separate `beeper-ci.yml` GitHub Actions workflow. +It runs exclusively on the `beeper*` branches, in place of the other +CI workflows that upstream uses and that we have not removed from our +fork. Don't get confused between the two. + +## Rebase flow + +### Create PR + +Here we're upgrading to `v1.96.1`: + +``` +# Make a new branch from the upstream release, we do this so we can create a PR +# of Beeper -> upstream to run tests/confirm we're happy. +git checkout -f v1.96.1 +git checkout -b upstream-1.96.1 +git push -u beeper upstream-1.96.1 + +# Check out the base branch, pull any changes +git checkout beeper +git pull + +# Now create a new branch to rebase +git checkout -b beeper-1.96.1 +# And do the rebase +git rebase v1.96.1 +# fix any conflicts... + +# Push and make a PR from this branch to the upstream one created above +git push -u beeper beeper-1.96.1 +``` + +### Make release + +Once it's ready we just overwrite the `beeper` branch with the new one: + +``` +git checkout beeper-1.96.1 +git push --force beeper beeper +``` diff --git a/beeper/complete_release.sh b/beeper/complete_release.sh new file mode 100755 index 0000000000..1d612ff350 --- /dev/null +++ b/beeper/complete_release.sh @@ -0,0 +1,31 @@ +#!/usr/bin/env bash + +set -euo pipefail +source $(realpath $(dirname $0))/utils.sh + +BEEPER_REMOTE=$(get_beeper_remote) + +VERSION=${1:-} + +if [ -z "$VERSION" ]; then + echo >&2 "Must specify version!" + exit 1 +fi + +echo "Completing Synapse: Beeper Edition version $VERSION" +echo "WARNING: this script will DELETE the branch called: beeper" +read -p "Press enter to continue" + +UPSTREAM_BRANCH=upstream-$VERSION +BEEPER_BRANCH=beeper-$VERSION + +git checkout $BEEPER_BRANCH +git branch -D beeper +git checkout -b beeper +git push --force $BEEPER_REMOTE beeper + +# Cleanup +git branch -D $BEEPER_BRANCH +git push $BEEPER_REMOTE --delete $BEEPER_BRANCH +git branch -D $UPSTREAM_BRANCH +git push $BEEPER_REMOTE --delete $UPSTREAM_BRANCH diff --git a/beeper/prepare_release.sh b/beeper/prepare_release.sh new file mode 100755 index 0000000000..ebd99537c3 --- /dev/null +++ b/beeper/prepare_release.sh @@ -0,0 +1,43 @@ +#!/usr/bin/env bash + +set -euo pipefail +source $(realpath $(dirname $0))/utils.sh + +BEEPER_REMOTE=$(get_beeper_remote) + +VERSION=${1:-} + +if [ -z "$VERSION" ]; then + echo >&2 "Must specify version!" + exit 1 +fi + +STARTING_BRANCH=$(git branch --show-current) + +echo "Preparing Synapse: Beeper Edition version $VERSION" +echo "WARNING: this script will rebase on top of the CURRENT BRANCH: $STARTING_BRANCH" +read -p "Press enter to continue" + +TAG=v$VERSION +UPSTREAM_BRANCH=upstream-$VERSION +BEEPER_BRANCH=beeper-$VERSION + +# Checkout the tag, create upstream branch, push it +echo "Setup branch $UPSTREAM_BRANCH" +git checkout -f $TAG +git checkout -b $UPSTREAM_BRANCH +git push -u $BEEPER_REMOTE $UPSTREAM_BRANCH + +# Switch back to our starting branch, create new version branch from it +echo "Setup branch $BEEPER_BRANCH" +git checkout $STARTING_BRANCH +git checkout -b $BEEPER_BRANCH + +# And rebase against upstream, applying only our Beeper commits +echo "Initiate rebase..." +git rebase $UPSTREAM_BRANCH || read -p "Rebase was a mess, press enter once you fix it" + +git push -u $BEEPER_REMOTE $BEEPER_BRANCH + +echo "OK we done!" +echo "Go HERE and make the PR: https://github.com/beeper/synapse/compare/upstream-$VERSION...beeper-$VERSION?expand=1" diff --git a/beeper/utils.sh b/beeper/utils.sh new file mode 100644 index 0000000000..ba7573dc16 --- /dev/null +++ b/beeper/utils.sh @@ -0,0 +1,23 @@ +function get_upstream_remote() { + for remote in $(git remote); do + url=$(git remote get-url $remote) + if [ "$url" = "git@github.com:element-hq/synapse.git" ]; then + echo $remote + return 0 + fi + done + echo >&2 "No upstream remote found (looking for URL: git@github.com:element-hq/synapse.git)" + return 1 +} + +function get_beeper_remote() { + for remote in $(git remote); do + url=$(git remote get-url $remote) + if [ "$url" = "git@github.com:beeper/synapse.git" ]; then + echo $remote + return 0 + fi + done + echo >&2 "No upstream remote found (looking for URL: git@github.com:beeper/synapse.git)" + return 1 +} diff --git a/docker/Dockerfile b/docker/Dockerfile index 30c6b670b6..1baa73f1d0 100644 --- a/docker/Dockerfile +++ b/docker/Dockerfile @@ -93,6 +93,9 @@ COPY --from=requirements /synapse/requirements.txt /synapse/ RUN --mount=type=cache,target=/root/.cache/uv \ uv pip install --prefix="/install" --no-deps -r /synapse/requirements.txt +# Beeper: install http antispam +RUN pip install --prefix="/install" --no-deps --no-warn-script-location 'synapse-http-antispam>=0.5.1' + # Copy over the rest of the synapse source code. COPY synapse /synapse/synapse/ COPY rust /synapse/rust/ diff --git a/pyproject.toml b/pyproject.toml index 172b1c3aec..3241558114 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -371,6 +371,7 @@ ignore = [ "B023", "E501", "E731", + "G004", ] select = [ # pycodestyle diff --git a/rust/src/events/constants.rs b/rust/src/events/constants.rs index 51f3b572f9..67dd109f21 100644 --- a/rust/src/events/constants.rs +++ b/rust/src/events/constants.rs @@ -80,6 +80,8 @@ pub mod unsigned_field { pub const TRANSACTION_ID: &str = "transaction_id"; /// Unsigned field: org.matrix.msc4140.delay_id pub const DELAY_ID: &str = "org.matrix.msc4140.delay_id"; + /// Unsigned field: Beeper homeserver stream ordering + pub const BEEPER_HS_ORDER: &str = "com.beeper.hs.order"; /// Unsigned field: membership (MSC4115) pub const MEMBERSHIP: &str = "membership"; /// Unsigned field: msc4354_sticky_duration_ttl_ms (MSC4354) diff --git a/rust/src/events/internal_metadata.rs b/rust/src/events/internal_metadata.rs index 269a5cc55e..3614d1a884 100644 --- a/rust/src/events/internal_metadata.rs +++ b/rust/src/events/internal_metadata.rs @@ -651,7 +651,7 @@ impl EventInternalMetadata { } #[getter] - fn get_stream_ordering(&self) -> PyResult> { + pub fn get_stream_ordering(&self) -> PyResult> { Ok(self.read_inner()?.stream_ordering) } #[setter] diff --git a/rust/src/events/serialize.rs b/rust/src/events/serialize.rs index 77a42ce5d2..5b81418cfa 100644 --- a/rust/src/events/serialize.rs +++ b/rust/src/events/serialize.rs @@ -467,6 +467,14 @@ fn serialize_event_value( } } + // Beeper: include the internal stream ordering as an HS order unsigned hint. + if let Some(stream_ordering) = event.internal_metadata.get_stream_ordering()? { + unsigned_mut(&mut d)?.insert( + unsigned_field::BEEPER_HS_ORDER.to_owned(), + Value::Number(Number::from(stream_ordering.get())), + ); + } + // Strip invite/knock room state unless requested. if !config.include_stripped_room_state { let unsigned = unsigned_mut(&mut d)?; diff --git a/rust/src/push/base_rules.rs b/rust/src/push/base_rules.rs index af7e6c83fb..15d23ea454 100644 --- a/rust/src/push/base_rules.rs +++ b/rust/src/push/base_rules.rs @@ -83,13 +83,70 @@ pub const BASE_APPEND_OVERRIDE_RULES: &[PushRule] = &[ default: true, default_enabled: true, }, + // Disable notifications for auto-accepted room invites + // NOTE: this rule must be a higher prio than .m.rule.invite_for_me because + // that will also match the same events. PushRule { - rule_id: Cow::Borrowed("global/override/.m.rule.suppress_notices"), + rule_id: Cow::Borrowed("global/override/.com.beeper.suppress_auto_invite"), + priority_class: 5, + conditions: Cow::Borrowed(&[ + Condition::Known(KnownCondition::EventMatch(EventMatchCondition { + key: Cow::Borrowed("type"), + pattern: Cow::Borrowed("m.room.member"), + })), + Condition::Known(KnownCondition::EventMatch(EventMatchCondition { + key: Cow::Borrowed("content.membership"), + pattern: Cow::Borrowed("invite"), + })), + Condition::Known(KnownCondition::EventMatchType(EventMatchTypeCondition { + key: Cow::Borrowed("state_key"), + pattern_type: Cow::Borrowed(&EventMatchPatternType::UserId), + })), + Condition::Known(KnownCondition::EventPropertyIs(EventPropertyIsCondition { + key: Cow::Borrowed("content.fi\\.mau\\.will_auto_accept"), + value: Cow::Borrowed(&SimpleJsonValue::Bool(true)), + })), + ]), + actions: Cow::Borrowed(&[]), + default: true, + default_enabled: true, + }, + // We don't want to notify on edits. Not only can this be confusing in real + // time (2 notifications, one message) but it's especially confusing + // if a bridge needs to edit a previously backfilled message. + PushRule { + rule_id: Cow::Borrowed("global/override/.com.beeper.suppress_edits"), priority_class: 5, conditions: Cow::Borrowed(&[Condition::Known(KnownCondition::EventMatch( EventMatchCondition { - key: Cow::Borrowed("content.msgtype"), - pattern: Cow::Borrowed("m.notice"), + key: Cow::Borrowed("content.m\\.relates_to.rel_type"), + pattern: Cow::Borrowed("m.replace"), + }, + ))]), + actions: Cow::Borrowed(&[]), + default: true, + default_enabled: true, + }, + PushRule { + rule_id: Cow::Borrowed("global/override/.com.beeper.suppress_send_message_status"), + priority_class: 5, + conditions: Cow::Borrowed(&[Condition::Known(KnownCondition::EventMatch( + EventMatchCondition { + key: Cow::Borrowed("type"), + pattern: Cow::Borrowed("com.beeper.message_send_status"), + }, + ))]), + actions: Cow::Borrowed(&[]), + default: true, + default_enabled: true, + }, + PushRule { + rule_id: Cow::Borrowed("global/override/.com.beeper.suppress_power_levels"), + priority_class: 5, + conditions: Cow::Borrowed(&[Condition::Known(KnownCondition::EventMatch( + EventMatchCondition { + key: Cow::Borrowed("type"), + pattern: Cow::Borrowed("m.room.power_levels"), }, ))]), actions: Cow::Borrowed(&[]), @@ -254,19 +311,6 @@ pub const BASE_APPEND_OVERRIDE_RULES: &[PushRule] = &[ default: true, default_enabled: true, }, - PushRule { - rule_id: Cow::Borrowed("global/override/.m.rule.reaction"), - priority_class: 5, - conditions: Cow::Borrowed(&[Condition::Known(KnownCondition::EventMatch( - EventMatchCondition { - key: Cow::Borrowed("type"), - pattern: Cow::Borrowed("m.reaction"), - }, - ))]), - actions: Cow::Borrowed(&[]), - default: true, - default_enabled: true, - }, PushRule { rule_id: Cow::Borrowed("global/override/.m.rule.room.server_acl"), priority_class: 5, @@ -352,6 +396,22 @@ pub const BASE_APPEND_POSTCONTENT_RULES: &[PushRule] = &[ ]; pub const BASE_APPEND_UNDERRIDE_RULES: &[PushRule] = &[ + // Beeper change: this rule is moved down from override. This means room + // rules take precedence, so if you enable bot notifications (by modifying + // this rule) notifications will not be sent for muted rooms. + PushRule { + rule_id: Cow::Borrowed("global/underride/.m.rule.suppress_notices"), + priority_class: 1, + conditions: Cow::Borrowed(&[Condition::Known(KnownCondition::EventMatch( + EventMatchCondition { + key: Cow::Borrowed("content.msgtype"), + pattern: Cow::Borrowed("m.notice"), + }, + ))]), + actions: Cow::Borrowed(&[]), + default: true, + default_enabled: true, + }, PushRule { rule_id: Cow::Borrowed("global/underride/.m.rule.call"), priority_class: 1, @@ -702,6 +762,32 @@ pub const BASE_APPEND_UNDERRIDE_RULES: &[PushRule] = &[ default: true, default_enabled: true, }, + // Enable notifications for reactions to your own messages *in rooms with less + // than 20 members*. + PushRule { + rule_id: Cow::Borrowed("global/underride/.com.beeper.reaction"), + priority_class: 1, + conditions: Cow::Borrowed(&[ + Condition::Known(KnownCondition::EventMatch(EventMatchCondition { + key: Cow::Borrowed("type"), + pattern: Cow::Borrowed("m.reaction"), + })), + Condition::Known(KnownCondition::RoomMemberCount { + is: Some(Cow::Borrowed("<20")), + }), + Condition::Known(KnownCondition::RelatedEventMatchType( + RelatedEventMatchTypeCondition { + key: Cow::Borrowed("sender"), + pattern_type: Cow::Borrowed(&EventMatchPatternType::UserId), + rel_type: Cow::Borrowed("m.annotation"), + include_fallbacks: None, + }, + )), + ]), + actions: Cow::Borrowed(&[Action::Notify, HIGHLIGHT_FALSE_ACTION]), + default: true, + default_enabled: true, + }, PushRule { rule_id: Cow::Borrowed("global/underride/.org.matrix.msc3930.rule.poll_start_one_to_one"), priority_class: 1, diff --git a/synapse/api/constants.py b/synapse/api/constants.py index 48c6f95bd1..321d141776 100644 --- a/synapse/api/constants.py +++ b/synapse/api/constants.py @@ -138,6 +138,7 @@ class LoginType: SSO: Final = "m.login.sso" DUMMY: Final = "m.login.dummy" REGISTRATION_TOKEN: Final = "m.login.registration_token" + JWT: Final = "org.matrix.login.jwt" # This is used in the `type` parameter for /register when called by @@ -380,6 +381,7 @@ class ReceiptTypes: READ: Final = "m.read" READ_PRIVATE: Final = "m.read.private" FULLY_READ: Final = "m.fully_read" + BEEPER_INBOX_DONE: Final = "com.beeper.inbox.done" class PublicRoomsFilterFields: diff --git a/synapse/api/errors.py b/synapse/api/errors.py index bce77ffba1..3bbac45c4d 100644 --- a/synapse/api/errors.py +++ b/synapse/api/errors.py @@ -130,6 +130,9 @@ class Codes(str, Enum): AS_FEDPROXY_CONNECTION_FAILED = "IO.ELEMENT.MSC4512.M_FEDPROXY_CONNECTION_FAILED" AS_FEDPROXY_DESTINATION_DENIED = "IO.ELEMENT.MSC4512.M_FEDPROXY_DESTINATION_DENIED" + # Beeper: account data compare-and-swap via com.beeper.expect_revision_id + EXPECTED_REVISION_ID_MISMATCH = "COM.BEEPER.REVISION_ID_MISMATCH" + # Attempt to send a second annotation with the same event type & annotation key # MSC2677 DUPLICATE_ANNOTATION = "M_DUPLICATE_ANNOTATION" diff --git a/synapse/app/generic_worker.py b/synapse/app/generic_worker.py index 6014d328ba..dbadc07793 100644 --- a/synapse/app/generic_worker.py +++ b/synapse/app/generic_worker.py @@ -63,6 +63,7 @@ ApplicationServiceTransactionWorkerStore, ApplicationServiceWorkerStore, ) +from synapse.storage.databases.main.beeper import BeeperStore from synapse.storage.databases.main.censor_events import CensorEventsStore from synapse.storage.databases.main.client_ips import ClientIpWorkerStore from synapse.storage.databases.main.delayed_events import DelayedEventsStore @@ -86,6 +87,7 @@ from synapse.storage.databases.main.monthly_active_users import ( MonthlyActiveUsersWorkerStore, ) +from synapse.storage.databases.main.openid import OpenIdStore from synapse.storage.databases.main.presence import PresenceStore from synapse.storage.databases.main.profile import ProfileWorkerStore from synapse.storage.databases.main.purge_events import PurgeEventsStore @@ -171,6 +173,8 @@ class GenericWorkerStore( ExperimentalFeaturesStore, SlidingSyncStore, DelayedEventsStore, + BeeperStore, + OpenIdStore, ): # Properties that multiple storage classes define. Tell mypy what the # expected type is. diff --git a/synapse/config/experimental.py b/synapse/config/experimental.py index 2e3782392e..05fee71852 100644 --- a/synapse/config/experimental.py +++ b/synapse/config/experimental.py @@ -316,3 +316,8 @@ def read_config( # MSC4512: Delegating parts of the C-S and S-S API to application services self.msc4512_enabled: bool = experimental.get("msc4512_enabled", False) + + self.beeper_user_notification_counts_enabled = experimental.get( + "beeper_user_notification_counts_enabled", + False, + ) diff --git a/synapse/handlers/account_data.py b/synapse/handlers/account_data.py index c6168377ee..1002a4394d 100644 --- a/synapse/handlers/account_data.py +++ b/synapse/handlers/account_data.py @@ -104,7 +104,8 @@ async def _notify_modules( logger.exception("Failed to run module callback %s: %s", callback, e) async def add_account_data_to_room( - self, user_id: str, room_id: str, account_data_type: str, content: JsonDict + self, user_id: str, room_id: str, account_data_type: str, content: JsonDict, + expected_revision_id: str | None = None, ) -> int: """Add some account_data to a room for a user. @@ -113,13 +114,15 @@ async def add_account_data_to_room( room_id: The room to add a tag for. account_data_type: The type of account_data to add. content: A json object to associate with the tag. + expected_revision_id: If set, only write if the stored content's + `com.beeper.revision_id` matches (compare-and-swap). Returns: The maximum stream ID. """ if self._instance_name in self._account_data_writers: max_stream_id = await self._store.add_account_data_to_room( - user_id, room_id, account_data_type, content + user_id, room_id, account_data_type, content, expected_revision_id ) self._notifier.on_new_event( @@ -136,6 +139,7 @@ async def add_account_data_to_room( room_id=room_id, account_data_type=account_data_type, content=content, + expected_revision_id=expected_revision_id, ) return response["max_stream_id"] @@ -181,7 +185,8 @@ async def remove_account_data_for_room( return response["max_stream_id"] async def add_account_data_for_user( - self, user_id: str, account_data_type: str, content: JsonDict + self, user_id: str, account_data_type: str, content: JsonDict, + expected_revision_id: str | None = None, ) -> int: """Add some global account_data for a user. @@ -189,6 +194,8 @@ async def add_account_data_for_user( user_id: The user to add some account data for. account_data_type: The type of account_data to add. content: The content json dictionary. + expected_revision_id: If set, only write if the stored content's + `com.beeper.revision_id` matches (compare-and-swap). Returns: The maximum stream ID. @@ -196,7 +203,7 @@ async def add_account_data_for_user( if self._instance_name in self._account_data_writers: max_stream_id = await self._store.add_account_data_for_user( - user_id, account_data_type, content + user_id, account_data_type, content, expected_revision_id ) self._notifier.on_new_event( @@ -212,6 +219,7 @@ async def add_account_data_for_user( user_id=user_id, account_data_type=account_data_type, content=content, + expected_revision_id=expected_revision_id, ) return response["max_stream_id"] diff --git a/synapse/handlers/auth.py b/synapse/handlers/auth.py index 47f83d4aad..08f998edf0 100644 --- a/synapse/handlers/auth.py +++ b/synapse/handlers/auth.py @@ -422,6 +422,10 @@ async def _get_available_ui_auth_types(self, user: UserID) -> Iterable[str]: ): ui_auth_types.add(LoginType.SSO) + # If JWT is enabled, allow user to re-authenticate with one + if self.hs.config.jwt.jwt_enabled: + ui_auth_types.add(LoginType.JWT) + return ui_auth_types def get_enabled_auth_types(self) -> Iterable[str]: diff --git a/synapse/handlers/device.py b/synapse/handlers/device.py index 46c7c7e60e..f0fd586d6b 100644 --- a/synapse/handlers/device.py +++ b/synapse/handlers/device.py @@ -210,6 +210,11 @@ def __init__(self, hs: "HomeServer"): desc="delete_stale_devices", func=self._delete_stale_devices, ) + # Beep: run this immediately since looping_call waits 24h after pod start + self.hs.run_as_background_process( + "delete_stale_devices", + func=self._delete_stale_devices, + ) async def _delete_stale_devices(self) -> None: """Background task that deletes devices which haven't been accessed for more than diff --git a/synapse/handlers/pagination.py b/synapse/handlers/pagination.py index 2bc7efeb5e..501e82daa9 100644 --- a/synapse/handlers/pagination.py +++ b/synapse/handlers/pagination.py @@ -19,9 +19,11 @@ # # import logging +import time from typing import TYPE_CHECKING, cast import attr +from prometheus_client import Histogram from twisted.python.failure import Failure @@ -33,6 +35,7 @@ from synapse.handlers.relations import BundledAggregations from synapse.handlers.worker_lock import NEW_EVENT_DURING_PURGE_LOCK_NAME from synapse.logging.opentracing import trace +from synapse.metrics import SERVER_NAME_LABEL from synapse.rest.admin._base import assert_user_is_admin from synapse.streams.config import PaginationConfig from synapse.types import ( @@ -55,6 +58,12 @@ logger = logging.getLogger(__name__) +purge_time = Histogram( + "room_purge_time", + "Time taken to purge rooms (sec)", + labelnames=[SERVER_NAME_LABEL], +) + # How many single event gaps we tolerate returning in a `/messages` response before we # backfill and try to fill in the history. This is an arbitrarily picked number so feel # free to tune it in the future. @@ -441,6 +450,7 @@ async def purge_room( room_id: room to be purged force: set true to skip checking for joined users. """ + purge_start = time.time() logger.info("starting purge room_id=%s force=%s", room_id, force) async with self._worker_locks.acquire_multi_read_write_lock( @@ -463,6 +473,10 @@ async def purge_room( await self._storage_controllers.purge_events.purge_room(room_id) + purge_end = time.time() + purge_time.labels(**{SERVER_NAME_LABEL: self.server_name}).observe( + purge_end - purge_start + ) logger.info("purge complete for room_id %s", room_id) @trace diff --git a/synapse/handlers/read_marker.py b/synapse/handlers/read_marker.py index 3f3b9e6d8b..706df46820 100644 --- a/synapse/handlers/read_marker.py +++ b/synapse/handlers/read_marker.py @@ -24,6 +24,7 @@ from synapse.api.constants import ReceiptTypes from synapse.api.errors import SynapseError +from synapse.types import JsonDict from synapse.util.async_helpers import Linearizer if TYPE_CHECKING: @@ -46,6 +47,7 @@ async def received_client_read_marker( user_id: str, event_id: str, allow_backward: bool = False, + extra_content: JsonDict | None = None, ) -> None: """Updates the read marker for a given user in a given room if the event ID given is ahead in the stream relative to the current read marker. @@ -77,7 +79,7 @@ async def received_client_read_marker( should_update = event_ordering > old_event_ordering if should_update: - content = {"event_id": event_id} + content = {"event_id": event_id, **(extra_content or {})} await self.account_data_handler.add_account_data_to_room( user_id, room_id, ReceiptTypes.FULLY_READ, content ) diff --git a/synapse/handlers/receipts.py b/synapse/handlers/receipts.py index 5ced086ef2..ea777596fd 100644 --- a/synapse/handlers/receipts.py +++ b/synapse/handlers/receipts.py @@ -190,6 +190,7 @@ async def received_client_receipt( user_id: UserID, event_id: str, thread_id: str | None, + extra_content: JsonDict | None = None, ) -> None: """Called when a client tells us a local user has read up to the given event_id in the room. @@ -206,7 +207,7 @@ async def received_client_receipt( user_id=user_id.to_string(), event_ids=[event_id], thread_id=thread_id, - data={"ts": int(self.clock.time_msec())}, + data={"ts": int(self.clock.time_msec()), **(extra_content or {})}, ) is_new = await self._handle_new_receipts([receipt]) diff --git a/synapse/handlers/relations.py b/synapse/handlers/relations.py index dee4746dcb..443453d7be 100644 --- a/synapse/handlers/relations.py +++ b/synapse/handlers/relations.py @@ -430,9 +430,16 @@ async def _get_threads_for_events( return results - @trace + # Beeper: bundled aggregations aren't used in Beeper clients and are a + # wasted calculation. async def get_bundled_aggregations( self, filtered_events: Iterable[FilteredEvent], user_id: str + ) -> dict[str, BundledAggregations]: + return {} + + @trace + async def _orig_get_bundled_aggregations( + self, filtered_events: Iterable[FilteredEvent], user_id: str ) -> dict[str, BundledAggregations]: """Generate bundled aggregations for events. diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py index 99a3396c1d..afa564200b 100644 --- a/synapse/handlers/room.py +++ b/synapse/handlers/room.py @@ -27,6 +27,7 @@ import math import random import string +import time from collections import OrderedDict from collections.abc import Mapping from http import HTTPStatus @@ -39,6 +40,7 @@ ) import attr +from prometheus_client import Histogram import synapse.events.snapshot from synapse.api.constants import ( @@ -75,6 +77,7 @@ copy_and_fixup_power_levels_contents, ) from synapse.handlers.relations import BundledAggregations +from synapse.metrics import SERVER_NAME_LABEL from synapse.rest.admin._base import assert_user_is_admin from synapse.streams import EventSource from synapse.types import ( @@ -113,6 +116,23 @@ FIVE_MINUTES_IN_MS = 5 * 60 * 1000 +shutdown_time = Histogram( + "room_shutdown_time", + "Time taken to shutdown rooms (sec)", + labelnames=[SERVER_NAME_LABEL], +) +shutdown_kick_count = Histogram( + "room_shutdown_kick_count", + "Number of users successfully kicked while shutting down a room", + labelnames=[SERVER_NAME_LABEL], +) +shutdown_failed_kick_count = Histogram( + "room_shutdown_failed_kick_count", + "Number of users that were failed to be kicked while shutting down a room", + labelnames=[SERVER_NAME_LABEL], +) + + @attr.s(slots=True, frozen=True, auto_attribs=True) class EventContext: events_before: list[FilteredEvent] @@ -159,7 +179,8 @@ def __init__(self, hs: "HomeServer"): "history_visibility": HistoryVisibility.SHARED, "original_invitees_have_ops": True, "guest_can_join": True, - "power_level_content_override": {"invite": 0}, + # Beeper change: don't allow redactions by anyone in DM chats + "power_level_content_override": {"invite": 0, "redact": 1000}, }, RoomCreationPreset.PUBLIC_CHAT: { "join_rules": JoinRules.PUBLIC, @@ -433,6 +454,9 @@ async def _upgrade_room( additional_creators, ) + # Beeper: clear out any push actions and summaries for this room + await self.store.beeper_cleanup_tombstoned_room(old_room_id) + return new_room_id async def _update_upgraded_room_pls( @@ -2491,6 +2515,7 @@ async def shutdown_room( else: logger.info("Shutting down room %r", room_id) + shutdown_start = time.time() users = await self.store.get_local_users_related_to_room(room_id) for user_id, membership in users: # If the user is not in the room (or is banned), nothing to do. @@ -2587,4 +2612,15 @@ async def shutdown_room( else: result["local_aliases"] = [] + shutdown_end = time.time() + shutdown_kick_count.labels(**{SERVER_NAME_LABEL: self.hs.hostname}).observe( + len(result["kicked_users"]) + ) + shutdown_failed_kick_count.labels( + **{SERVER_NAME_LABEL: self.hs.hostname} + ).observe(len(result["failed_to_kick_users"])) + shutdown_time.labels(**{SERVER_NAME_LABEL: self.hs.hostname}).observe( + shutdown_end - shutdown_start + ) + return result diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py index d635465e5c..1ea3409ed4 100644 --- a/synapse/handlers/sync.py +++ b/synapse/handlers/sync.py @@ -241,6 +241,7 @@ class SyncConfig: # FIXME(unstable_state_after): Remove support for the unstable identifiers after 2027-09-01 # (to allow some time for the ecosystem to adapt to the stable identifiers) use_unstable_state_after_name: bool = False + beeper_previews: bool = False @attr.s(slots=True, frozen=True, auto_attribs=True) @@ -276,6 +277,7 @@ class JoinedSyncResult: unread_thread_notifications: JsonDict summary: JsonDict | None unread_count: int + preview: JsonDict | None def __bool__(self) -> bool: """Make the result appear empty if there are no updates. This is used @@ -287,6 +289,7 @@ def __bool__(self) -> bool: or self.ephemeral or self.account_data or self.sticky + or self.preview # nb the notification count does not, er, count: if there's nothing # else in the result, we don't need to send it. ) @@ -3475,6 +3478,8 @@ async def _generate_room_entry( } ) + user_id = sync_result_builder.sync_config.user.to_string() + # Note: `batch` can be both empty and limited here in the case where # `_load_filtered_recents` can't find any events the user should see # (e.g. due to having ignored the sender of the last 50 events). @@ -3484,7 +3489,6 @@ async def _generate_room_entry( # newly joined room, unless either a) they've joined before or b) the # tag was added by synapse e.g. for server notice rooms. if full_state: - user_id = sync_result_builder.sync_config.user.to_string() tags = await self.store.get_tags_for_room(user_id, room_id) # If there aren't any tags, don't send the empty tags list down @@ -3611,8 +3615,44 @@ async def _generate_room_entry( summary=summary, unread_count=0, sticky=sticky_events, + preview=None, ) + # Only generate previews if we have new events that would change it + if batch.events and sync_config.beeper_previews: + preview = ( + await self.store.beeper_preview_event_for_room_id_and_user_id( + room_id=room_id, user_id=user_id, to_key=now_token.room_key + ) + ) + + if preview: + preview_event_id, preview_origin_server_ts = preview + room_sync.preview = { + "event_id": preview_event_id, + "origin_server_ts": preview_origin_server_ts, + } + + # Check if we already have the event in the batch, in which + # case we needn't add it here. No point in checking state as + # we don't preview state events. + for ev in batch.events: + if ev.event.event_id == preview_event_id: + break + else: + preview_event = await self.store.get_event( + preview_event_id, + allow_none=True, + ) + if preview_event is not None: + room_sync.preview["event"] = FilteredEvent( + event=preview_event, + membership=None, + ) + else: + # This should never happen! + logger.warning("Beeper preview is missing! roomID=%s", room_id) + if room_sync or always_include: notifs = await self.unread_notifs_for_room_id(room_id, sync_config) diff --git a/synapse/handlers/ui_auth/checkers.py b/synapse/handlers/ui_auth/checkers.py index a0097dbc96..6846b3b272 100644 --- a/synapse/handlers/ui_auth/checkers.py +++ b/synapse/handlers/ui_auth/checkers.py @@ -27,6 +27,7 @@ from synapse.api.constants import LoginType from synapse.api.errors import Codes, LoginError, SynapseError +from synapse.types import UserID from synapse.util.json import json_decoder if TYPE_CHECKING: @@ -321,6 +322,87 @@ async def check_auth(self, authdict: dict, clientip: str) -> Any: ) +class JwtAuthChecker(UserInteractiveAuthChecker): + AUTH_TYPE = LoginType.JWT + + def __init__(self, hs: "HomeServer"): + super().__init__(hs) + self.hs = hs + + def is_enabled(self) -> bool: + return bool(self.hs.config.jwt.jwt_enabled) + + async def check_auth(self, authdict: dict, clientip: str) -> Any: + token = authdict.get("token", None) + if token is None: + raise LoginError( + 403, "Token field for JWT is missing", errcode=Codes.FORBIDDEN + ) + + from authlib.jose import JsonWebToken, JWTClaims + from authlib.jose.errors import BadSignatureError, InvalidClaimError, JoseError + + jwt = JsonWebToken([self.hs.config.jwt.jwt_algorithm]) + claim_options = {} + if self.hs.config.jwt.jwt_issuer is not None: + claim_options["iss"] = { + "value": self.hs.config.jwt.jwt_issuer, + "essential": True, + } + if self.hs.config.jwt.jwt_audiences is not None: + claim_options["aud"] = { + "values": self.hs.config.jwt.jwt_audiences, + "essential": True, + } + + try: + claims = jwt.decode( + token, + key=self.hs.config.jwt.jwt_secret, + claims_cls=JWTClaims, + claims_options=claim_options, + ) + except BadSignatureError: + # We handle this case separately to provide a better error message + raise LoginError( + 403, + "JWT validation failed: Signature verification failed", + errcode=Codes.FORBIDDEN, + ) + except JoseError as e: + # A JWT error occurred, return some info back to the client. + raise LoginError( + 403, + "JWT validation failed: %s" % (str(e),), + errcode=Codes.FORBIDDEN, + ) + + try: + claims.validate(leeway=120) # allows 2 min of clock skew + + # Enforce the old behavior which is rolled out in productive + # servers: if the JWT contains an 'aud' claim but none is + # configured, the login attempt will fail + if claims.get("aud") is not None: + if ( + self.hs.config.jwt.jwt_audiences is None + or len(self.hs.config.jwt.jwt_audiences) == 0 + ): + raise InvalidClaimError("aud") + except JoseError as e: + raise LoginError( + 403, + "JWT validation failed: %s" % (str(e),), + errcode=Codes.FORBIDDEN, + ) + + user = claims.get(self.hs.config.jwt.jwt_subject_claim, None) + if user is None: + raise LoginError(403, "Invalid JWT", errcode=Codes.FORBIDDEN) + + return UserID(user, self.hs.hostname).to_string() + + INTERACTIVE_AUTH_CHECKERS: Sequence[type[UserInteractiveAuthChecker]] = [ DummyAuthChecker, TermsAuthChecker, @@ -328,5 +410,6 @@ async def check_auth(self, authdict: dict, clientip: str) -> Any: EmailIdentityAuthChecker, MsisdnAuthChecker, RegistrationTokenAuthChecker, + JwtAuthChecker, ] """A list of UserInteractiveAuthChecker classes""" diff --git a/synapse/logging/__init__.py b/synapse/logging/__init__.py index 4058561e5e..009f71edf6 100644 --- a/synapse/logging/__init__.py +++ b/synapse/logging/__init__.py @@ -23,13 +23,20 @@ from synapse.logging._remote import RemoteHandler from synapse.logging._terse_json import ( + BeeperTerseJsonFormatter, GcpJsonFormatter, JsonFormatter, TerseJsonFormatter, ) # These are imported to allow for nicer logging configuration files. -__all__ = ["RemoteHandler", "JsonFormatter", "TerseJsonFormatter", "GcpJsonFormatter"] +__all__ = [ + "RemoteHandler", + "JsonFormatter", + "TerseJsonFormatter", + "BeeperTerseJsonFormatter", + "GcpJsonFormatter", +] # Debug logger for https://github.com/matrix-org/synapse/issues/9533 etc issue9533_logger = logging.getLogger("synapse.9533_debug") diff --git a/synapse/logging/_terse_json.py b/synapse/logging/_terse_json.py index afa3288809..ec59e47513 100644 --- a/synapse/logging/_terse_json.py +++ b/synapse/logging/_terse_json.py @@ -122,3 +122,17 @@ def format(self, record: logging.LogRecord) -> str: } return _encoder.encode(event) + + +class BeeperTerseJsonFormatter(JsonFormatter): + def format(self, record: logging.LogRecord) -> str: + event = { + "message": record.getMessage(), + "namespace": record.name, + "level": record.levelname.lower(), + "time": round(record.created, 2), + } + if record.exc_info and record.exc_info[0]: + event["exc_traceback"] = self.formatException(record.exc_info) + + return self._format(record, event) diff --git a/synapse/push/bulk_push_rule_evaluator.py b/synapse/push/bulk_push_rule_evaluator.py index 4efbce1029..456524f2de 100644 --- a/synapse/push/bulk_push_rule_evaluator.py +++ b/synapse/push/bulk_push_rule_evaluator.py @@ -89,36 +89,36 @@ SENTINEL = object() -def _should_count_as_unread(event: EventBase, context: EventContext) -> bool: +def _should_count_as_unread( + event: EventBase, + context: EventContext, + non_bot_room_members_count: int, + current_user: str, + related_events: dict[str, dict[str, Any]], +) -> bool: # Exclude rejected and soft-failed events. if context.rejected or event.internal_metadata.is_soft_failed(): return False - # Exclude notices. - if ( - not event.is_state() - and event.type == EventTypes.Message - and event.content.get("msgtype") == "m.notice" - ): - return False - # Exclude edits. relates_to = relation_from_event(event) if relates_to and relates_to.rel_type == RelationTypes.REPLACE: return False - # Mark events that have a non-empty string body as unread. - body = event.content.get("body") - if isinstance(body, str) and body: - return True - - # Mark some state events as unread. - if event.is_state() and event.type in STATE_EVENT_TYPES_TO_MARK_UNREAD: - return True - - # Mark encrypted events as unread. - if not event.is_state() and event.type == EventTypes.Encrypted: - return True + # Mark encrypted and plain text messages events as unread. + if not event.is_state(): + if event.type == EventTypes.Encrypted: + return True + elif event.type == EventTypes.Message: + body = event.content.get("body") + return isinstance(body, str) and bool(body) + # Beeper: We want reactions to only count as unread if they're reactions to the current user in rooms that + # have fewer than 20 users. + elif event.type == "m.reaction" and related_events.get("m.annotation"): + return ( + related_events["m.annotation"]["sender"] == current_user + and non_bot_room_members_count < 20 + ) return False @@ -382,13 +382,6 @@ async def _action_for_event_by_user( # (historical messages persisted in reverse-chronological order). return - # Disable counting as unread unless the experimental configuration is - # enabled, as it can cause additional (unwanted) rows to be added to the - # event_push_actions table. - count_as_unread = False - if self.hs.config.experimental.msc2654_enabled: - count_as_unread = _should_count_as_unread(event, context) - rules_by_user = await self._get_rules_for_event(event) if not rules_by_user: # There are no local users (other than the sender) who could be @@ -396,6 +389,7 @@ async def _action_for_event_by_user( return actions_by_user: dict[str, Collection[Mapping | str]] = {} + count_as_unread_by_user: dict[str, bool] = {} # Gather a bunch of info in parallel. # @@ -505,12 +499,19 @@ async def _action_for_event_by_user( if not isinstance(display_name, str): display_name = None - if count_as_unread: - # Add an element for the current user if the event needs to be marked as - # unread, so that add_push_actions_to_staging iterates over it. - # If the event shouldn't be marked as unread but should notify the - # current user, it'll be added to the dict later. - actions_by_user[uid] = [] + # Beeper: Need to calculate this per user as whether it should count as unread or not + # depends on who the current user is. + if self.hs.config.experimental.msc2654_enabled: + count_as_unread_by_user[uid] = _should_count_as_unread( + event, context, room_member_count, uid, related_events + ) + + if count_as_unread_by_user[uid]: + # Add an element for the current user if the event needs to be marked as + # unread, so that add_push_actions_to_staging iterates over it. + # If the event shouldn't be marked as unread but should notify the + # current user, it'll be added to the dict later. + actions_by_user[uid] = [] msc4306_thread_subscription_state: bool | None = None if msc4306_thread_subscribers is not None: @@ -549,7 +550,7 @@ async def _action_for_event_by_user( await self.store.add_push_actions_to_staging( event.event_id, actions_by_user, - count_as_unread, + count_as_unread_by_user, thread_id, ) diff --git a/synapse/push/httppusher.py b/synapse/push/httppusher.py index ca63a99e3e..a08a4da840 100644 --- a/synapse/push/httppusher.py +++ b/synapse/push/httppusher.py @@ -143,6 +143,8 @@ def __init__(self, hs: "HomeServer", pusher_config: PusherConfig): pusher_config.app_id, pusher_config.pushkey, ) + # Beeper: Save this so we can pass this on to Sygnal as well + self.user_name = pusher_config.user_name # Validate that there's a URL and it is of the proper form. if "url" not in self.data: @@ -469,7 +471,7 @@ async def dispatch_push_event( rejected push keys otherwise. If this array is empty, the push fully succeeded. """ - priority = "low" + priority = "high" # Beep: always use high priority if ( event.type == EventTypes.Encrypted or tweaks.get("highlight") @@ -485,10 +487,14 @@ async def dispatch_push_event( content: JsonDict = { "event_id": event.event_id, "room_id": event.room_id, + "com.beeper.user_id": self.user_id, "prio": priority, } if not self.disable_badge_count: - content["counts"] = {"unread": badge} + content["counts"] = { + "unread": badge, + "com.beeper.server_type": "synapse", + } # event_id_only doesn't include the tweaks, so override them. tweaks = {} else: @@ -503,10 +509,12 @@ async def dispatch_push_event( "type": event.type, "sender": event.sender, "prio": priority, + "com.beeper.user_id": self.user_id, } if not self.disable_badge_count: content["counts"] = { "unread": badge, + "com.beeper.server_type": "synapse", } if event.type == "m.room.member" and event.is_state(): content["membership"] = event.content["membership"] @@ -540,7 +548,11 @@ async def _send_badge(self, badge: int) -> None: "id": "", "type": None, "sender": "", - "counts": {"unread": badge}, + "counts": { + "unread": badge, + "com.beeper.server_type": "synapse", + }, + "com.beeper.user_id": self.user_id, "devices": [ { "app_id": self.app_id, diff --git a/synapse/replication/http/account_data.py b/synapse/replication/http/account_data.py index 560973b916..770af4fc31 100644 --- a/synapse/replication/http/account_data.py +++ b/synapse/replication/http/account_data.py @@ -58,11 +58,14 @@ def __init__(self, hs: "HomeServer"): @staticmethod async def _serialize_payload( # type: ignore[override] - user_id: str, account_data_type: str, content: JsonDict + user_id: str, account_data_type: str, content: JsonDict, + expected_revision_id: str | None = None, ) -> JsonDict: - payload = { + payload: JsonDict = { "content": content, } + if expected_revision_id is not None: + payload["expected_revision_id"] = expected_revision_id return payload @@ -70,7 +73,8 @@ async def _handle_request( # type: ignore[override] self, request: Request, content: JsonDict, user_id: str, account_data_type: str ) -> tuple[int, JsonDict]: max_stream_id = await self.handler.add_account_data_for_user( - user_id, account_data_type, content["content"] + user_id, account_data_type, content["content"], + expected_revision_id=content.get("expected_revision_id"), ) return 200, {"max_stream_id": max_stream_id} @@ -138,11 +142,14 @@ def __init__(self, hs: "HomeServer"): @staticmethod async def _serialize_payload( # type: ignore[override] - user_id: str, room_id: str, account_data_type: str, content: JsonDict + user_id: str, room_id: str, account_data_type: str, content: JsonDict, + expected_revision_id: str | None = None, ) -> JsonDict: - payload = { + payload: JsonDict = { "content": content, } + if expected_revision_id is not None: + payload["expected_revision_id"] = expected_revision_id return payload @@ -155,7 +162,8 @@ async def _handle_request( # type: ignore[override] account_data_type: str, ) -> tuple[int, JsonDict]: max_stream_id = await self.handler.add_account_data_to_room( - user_id, room_id, account_data_type, content["content"] + user_id, room_id, account_data_type, content["content"], + expected_revision_id=content.get("expected_revision_id"), ) return 200, {"max_stream_id": max_stream_id} diff --git a/synapse/res/templates/_base.html b/synapse/res/templates/_base.html index 4b5cc7bcb6..b36d316a13 100644 --- a/synapse/res/templates/_base.html +++ b/synapse/res/templates/_base.html @@ -12,15 +12,7 @@
- {% if app_name == "Riot" %} - [Riot] - {% elif app_name == "Vector" %} - [Vector] - {% elif app_name == "Element" %} - [Element] - {% else %} - [matrix] - {% endif %} + [Beeper]
{% block body %}{% endblock %} diff --git a/synapse/res/templates/password_reset.html b/synapse/res/templates/password_reset.html index 1f267946c8..0b0c969554 100644 --- a/synapse/res/templates/password_reset.html +++ b/synapse/res/templates/password_reset.html @@ -2,9 +2,9 @@ {% block title %}Password reset{% endblock %} {% block body %} -

A password reset request has been received for your Matrix account. If this was you, please click the link below to confirm resetting your password:

+

A password reset request has been received for your Beeper account. If this was you, please click the link below to confirm resetting your password:

{{ link }} -

If this was not you, do not click the link above and instead contact your server administrator. Thank you.

+

If this was not you, do not click the link above and instead contact the Beeper Support team. Thank you.

{% endblock %} diff --git a/synapse/res/templates/password_reset_confirmation.html b/synapse/res/templates/password_reset_confirmation.html index fabb9a6ed5..6af2d5aa7c 100644 --- a/synapse/res/templates/password_reset_confirmation.html +++ b/synapse/res/templates/password_reset_confirmation.html @@ -8,7 +8,7 @@ -

You have requested to reset your Matrix account password. Click the link below to confirm this action.

+

You have requested to reset your Beeper account password. Click the link below to confirm this action.

If you did not mean to do this, please close this page and your password will not be changed.

diff --git a/synapse/rest/admin/__init__.py b/synapse/rest/admin/__init__.py index 2e106826ff..52145ef27f 100644 --- a/synapse/rest/admin/__init__.py +++ b/synapse/rest/admin/__init__.py @@ -283,6 +283,8 @@ def register_servlets(hs: "HomeServer", http_server: HttpServer) -> None: if hs.config.worker.worker_app is not None: UserRestServletV2Get(hs).register(http_server) + # Beep: except this one ;) + UserTokenRestServlet(hs).register(http_server) return auth_delegated = hs.config.mas.enabled diff --git a/synapse/rest/admin/rooms.py b/synapse/rest/admin/rooms.py index e47b6e9efe..6193adebcf 100644 --- a/synapse/rest/admin/rooms.py +++ b/synapse/rest/admin/rooms.py @@ -19,11 +19,13 @@ # # import logging +import time from http import HTTPStatus from typing import TYPE_CHECKING, cast import attr from immutabledict import immutabledict +from prometheus_client import Histogram from synapse.api.constants import Direction, EventTypes, JoinRules, Membership from synapse.api.errors import AuthError, Codes, NotFoundError, SynapseError @@ -50,6 +52,7 @@ ) from synapse.http.site import SynapseRequest from synapse.logging.opentracing import trace +from synapse.metrics import SERVER_NAME_LABEL from synapse.rest.admin._base import ( admin_patterns, assert_requester_is_admin, @@ -76,6 +79,12 @@ logger = logging.getLogger(__name__) +delete_time = Histogram( + "admin_room_delete_time", + "Time taken to delete rooms via the admin API (sec)", + labelnames=[SERVER_NAME_LABEL], +) + class AdminRoomHierarchy(RestServlet): """ @@ -406,13 +415,21 @@ async def on_GET( async def on_DELETE( self, request: SynapseRequest, room_id: str ) -> tuple[int, JsonDict]: - return await self._delete_room( + logger.info(f"[admin/rooms] deleting {room_id}") + start = time.time() + response = await self._delete_room( request, room_id, self.auth, self.room_shutdown_handler, self.pagination_handler, ) + end = time.time() + logger.info(f"[admin/rooms] deleting {room_id} took {end - start} seconds") + delete_time.labels(**{SERVER_NAME_LABEL: request.our_server_name}).observe( + end - start + ) + return response async def _delete_room( self, diff --git a/synapse/rest/client/account_data.py b/synapse/rest/client/account_data.py index cccc0473e8..fbefa3e62b 100644 --- a/synapse/rest/client/account_data.py +++ b/synapse/rest/client/account_data.py @@ -19,14 +19,20 @@ # # +import json import logging from typing import TYPE_CHECKING from synapse.api.constants import AccountDataTypes, ReceiptTypes from synapse.api.errors import AuthError, Codes, NotFoundError, SynapseError from synapse.http.server import HttpServer -from synapse.http.servlet import RestServlet, parse_json_object_from_request +from synapse.http.servlet import ( + RestServlet, + parse_json_object_from_request, + parse_string, +) from synapse.http.site import SynapseRequest +from synapse.rest.client.read_marker import ReadMarkerRestServlet from synapse.types import JsonDict, JsonMapping, RoomID from ._base import client_patterns @@ -84,6 +90,7 @@ async def on_PUT( _check_can_set_account_data_type(account_data_type) body = parse_json_object_from_request(request) + expected_revision_id = parse_string(request, "com.beeper.expect_revision_id") # If experimental support for MSC3391 is enabled, then providing an empty dict # as the value for an account data type should be functionally equivalent to @@ -95,7 +102,9 @@ async def on_PUT( ) return 200, {} - await self.handler.add_account_data_for_user(user_id, account_data_type, body) + await self.handler.add_account_data_for_user( + user_id, account_data_type, body, expected_revision_id + ) return 200, {} @@ -207,6 +216,7 @@ async def on_PUT( _check_can_set_account_data_type(account_data_type) body = parse_json_object_from_request(request) + expected_revision_id = parse_string(request, "com.beeper.expect_revision_id") # If experimental support for MSC3391 is enabled, then providing an empty dict # as the value for an account data type should be functionally equivalent to @@ -219,7 +229,7 @@ async def on_PUT( return 200, {} await self.handler.add_account_data_to_room( - user_id, room_id, account_data_type, body + user_id, room_id, account_data_type, body, expected_revision_id ) return 200, {} @@ -310,9 +320,113 @@ async def on_DELETE( return 200, {} +class RoomBeeperInboxStateServlet(RestServlet): + """ + PUT /user/{user_id}/rooms/{room_id}/beeper_inbox_state HTTP/1.1 + """ + + PATTERNS = list( + client_patterns( + "/com.beeper.inbox/user/(?P[^/]*)/rooms/(?P[^/]*)/inbox_state", + releases=(), # not in the matrix spec, only include under /unstable + ) + ) + + def __init__(self, hs: "HomeServer"): + super().__init__() + self.auth = hs.get_auth() + self.clock = hs.get_clock() + self.store = hs.get_datastores().main + self.handler = hs.get_account_data_handler() + self.read_marker_client = ReadMarkerRestServlet(hs) + + async def on_PUT( + self, request: SynapseRequest, user_id: str, room_id: str + ) -> tuple[int, JsonDict]: + requester = await self.auth.get_user_by_req(request) + if user_id != requester.user.to_string(): + raise AuthError(403, "Cannot add beeper inbox state for other users.") + + if not RoomID.is_valid(room_id): + raise SynapseError( + 400, + f"{room_id} is not a valid room ID", + Codes.INVALID_PARAM, + ) + + ts = self.clock.time_msec() + + body = parse_json_object_from_request(request) + + if "done" in body: + delta_ms = body["done"].get("at_delta") or 0 + done = {"updated_ts": ts, "at_ts": ts + delta_ms} + if "at_order" in body["done"]: + done["at_order"] = body["done"]["at_order"] + await self.handler.add_account_data_to_room( + user_id, room_id, "com.beeper.inbox.done", done + ) + logger.info( + f"SetBeeperDone done_delta_ms={delta_ms} at_order={body.get('at_order')}" + ) + + if "marked_unread" in body: + marked_unread = {"unread": body["marked_unread"], "ts": ts} + await self.handler.add_account_data_to_room( + user_id, room_id, "m.marked_unread", marked_unread + ) + logger.info(f"SetBeeperMarkedUnread marked_unread={body['marked_unread']}") + + if "read_markers" in body: + await self.read_marker_client.handle_read_marker( + room_id, body["read_markers"], requester + ) + logger.info( + f"SetBeeperReadMarkers read_markers={json.dumps(body['read_markers'])}" + ) + + return 200, {} + + +class BeeperInboxBatchArchiveServlet(RestServlet): + """ + PUT /com.beeper.inbox/batch_archive HTTP/1.1 + """ + + PATTERNS = list( + client_patterns( + "/com.beeper.inbox/batch_archive", + releases=(), # not in the matrix spec, only include under /unstable + ) + ) + + def __init__(self, hs: "HomeServer"): + super().__init__() + self.auth = hs.get_auth() + self.clock = hs.get_clock() + self.store = hs.get_datastores().main + self.handler = hs.get_account_data_handler() + + async def on_POST(self, request: SynapseRequest) -> tuple[int, JsonDict]: + requester = await self.auth.get_user_by_req(request) + ts = self.clock.time_msec() + body = parse_json_object_from_request(request) + + done = {"updated_ts": ts, "at_ts": ts} + for room_id in body["room_ids"]: + # TODO in transaction + await self.handler.add_account_data_to_room( + requester.user.to_string(), room_id, "com.beeper.inbox.done", done + ) + + return 200, {} + + def register_servlets(hs: "HomeServer", http_server: HttpServer) -> None: AccountDataServlet(hs).register(http_server) RoomAccountDataServlet(hs).register(http_server) + RoomBeeperInboxStateServlet(hs).register(http_server) + BeeperInboxBatchArchiveServlet(hs).register(http_server) if hs.config.experimental.msc3391_enabled: UnstableAccountDataServlet(hs).register(http_server) diff --git a/synapse/rest/client/notifications.py b/synapse/rest/client/notifications.py index ae3893d296..28cf0c662e 100644 --- a/synapse/rest/client/notifications.py +++ b/synapse/rest/client/notifications.py @@ -87,6 +87,7 @@ async def on_GET(self, request: SynapseRequest) -> tuple[int, JsonDict]: [ ReceiptTypes.READ, ReceiptTypes.READ_PRIVATE, + ReceiptTypes.BEEPER_INBOX_DONE, ], ) diff --git a/synapse/rest/client/read_marker.py b/synapse/rest/client/read_marker.py index 8e0f2a2e7a..9fb2fbba5f 100644 --- a/synapse/rest/client/read_marker.py +++ b/synapse/rest/client/read_marker.py @@ -27,7 +27,7 @@ from synapse.http.server import HttpServer from synapse.http.servlet import RestServlet, parse_json_object_from_request from synapse.http.site import SynapseRequest -from synapse.types import JsonDict +from synapse.types import JsonDict, Requester from ._base import client_patterns @@ -59,13 +59,18 @@ async def on_POST( self, request: SynapseRequest, room_id: str ) -> tuple[int, JsonDict]: requester = await self.auth.get_user_by_req(request) + body = parse_json_object_from_request(request) + return await self.handle_read_marker(room_id, body, requester) + # Beeper: The endpoint and underlying method are separated here so `inbox_state` + # can use the same function. + async def handle_read_marker( + self, room_id: str, body: JsonDict, requester: Requester + ) -> tuple[int, JsonDict]: await self.presence_handler.bump_presence_active_time( requester.user, requester.device_id ) - body = parse_json_object_from_request(request) - unrecognized_types = set(body.keys()) - self._known_receipt_types if self.config.experimental.msc4446_enabled: @@ -103,6 +108,7 @@ async def on_POST( user_id=requester.user.to_string(), event_id=event_id, allow_backward=allow_backward, + extra_content=body.get("com.beeper.fully_read.extra", None), ) else: await self.receipts_handler.received_client_receipt( @@ -112,6 +118,7 @@ async def on_POST( event_id=event_id, # Setting the thread ID is not possible with the /read_markers endpoint. thread_id=None, + extra_content=body.get("com.beeper.read.extra", None), ) return 200, {} diff --git a/synapse/rest/client/receipts.py b/synapse/rest/client/receipts.py index 949a1e64ad..52c876d2d5 100644 --- a/synapse/rest/client/receipts.py +++ b/synapse/rest/client/receipts.py @@ -57,6 +57,7 @@ def __init__(self, hs: "HomeServer"): ReceiptTypes.READ, ReceiptTypes.READ_PRIVATE, ReceiptTypes.FULLY_READ, + ReceiptTypes.BEEPER_INBOX_DONE, } async def on_POST( @@ -73,7 +74,7 @@ async def on_POST( f"Receipt type must be {', '.join(self._known_receipt_types)}", ) - body = parse_json_object_from_request(request) + body = parse_json_object_from_request(request, allow_empty_body=False) if self._msc4446_enabled: allow_backward = body.get("com.beeper.allow_backward", False) @@ -130,6 +131,7 @@ async def on_POST( user_id=requester.user.to_string(), event_id=event_id, allow_backward=allow_backward, + extra_content=body, ) else: await self.receipts_handler.received_client_receipt( @@ -138,6 +140,7 @@ async def on_POST( user_id=requester.user, event_id=event_id, thread_id=thread_id, + extra_content=body, ) return 200, {} diff --git a/synapse/rest/client/register.py b/synapse/rest/client/register.py index 6e4bb400af..b583f89739 100644 --- a/synapse/rest/client/register.py +++ b/synapse/rest/client/register.py @@ -369,10 +369,8 @@ def __init__(self, hs: "HomeServer"): ) async def on_GET(self, request: Request) -> tuple[int, JsonDict]: - if not self.hs.config.registration.enable_registration: - raise SynapseError( - 403, "Registration has been disabled", errcode=Codes.FORBIDDEN - ) + # Beeper: allow availability checks even when public registration is + # disabled. if self.inhibit_user_in_use_error: return 200, {"available": True} diff --git a/synapse/rest/client/sync.py b/synapse/rest/client/sync.py index bf4b2c660a..0feecbf5e3 100644 --- a/synapse/rest/client/sync.py +++ b/synapse/rest/client/sync.py @@ -174,6 +174,7 @@ async def on_GET(self, request: SynapseRequest) -> tuple[int, JsonDict]: ) filter_id = parse_string(request, "filter") full_state = parse_boolean(request, "full_state", default=False) + beeper_previews = parse_boolean(request, "beeper_previews", default=False) use_state_after = parse_boolean(request, "use_state_after", default=False) use_unstable_state_after_name = False @@ -186,13 +187,14 @@ async def on_GET(self, request: SynapseRequest) -> tuple[int, JsonDict]: logger.debug( "/sync: user=%r, timeout=%r, since=%r, " - "set_presence=%r, filter_id=%r, device_id=%r", + "set_presence=%r, filter_id=%r, device_id=%r, beeper_previews=%r", user, timeout, since, set_presence, filter_id, device_id, + beeper_previews, ) # Stream position of the last ignored users account data event for this user, @@ -218,6 +220,7 @@ async def on_GET(self, request: SynapseRequest) -> tuple[int, JsonDict]: last_ignore_accdata_streampos, use_state_after, use_unstable_state_after_name, + beeper_previews, ) if filter_id is None: @@ -256,6 +259,7 @@ async def on_GET(self, request: SynapseRequest) -> tuple[int, JsonDict]: device_id=device_id, use_state_after=use_state_after, use_unstable_state_after_name=use_unstable_state_after_name, + beeper_previews=beeper_previews, ) since_token = None @@ -679,6 +683,17 @@ async def encode_room( if self._msc2654_enabled: result["org.matrix.msc2654.unread_count"] = room.unread_count + if room.preview: + if "event" in room.preview: + room.preview["event"] = ( + await self._event_serializer.serialize_events( + [room.preview["event"]], + time_now, + config=serialize_options, + ) + )[0] + result["com.beeper.inbox.preview"] = room.preview + return result diff --git a/synapse/rest/well_known.py b/synapse/rest/well_known.py index b2e274e3f3..29c3b5fddc 100644 --- a/synapse/rest/well_known.py +++ b/synapse/rest/well_known.py @@ -50,6 +50,9 @@ async def get_well_known(self) -> JsonDict | None: result["m.identity_server"] = { "base_url": self._config.registration.default_identity_server } + else: + # Workaround for iOS expecting some value here + result["m.identity_server"] = {"base_url": ""} if self._config.server.extra_well_known_client_content: for ( diff --git a/synapse/storage/databases/main/__init__.py b/synapse/storage/databases/main/__init__.py index 9f8d4debbe..b04e874505 100644 --- a/synapse/storage/databases/main/__init__.py +++ b/synapse/storage/databases/main/__init__.py @@ -44,6 +44,7 @@ from .account_data import AccountDataStore from .appservice import ApplicationServiceStore, ApplicationServiceTransactionStore +from .beeper import BeeperStore from .cache import CacheInvalidationWorkerStore from .censor_events import CensorEventsStore from .client_ips import ClientIpWorkerStore @@ -165,6 +166,7 @@ class DataStore( TaskSchedulerWorkerStore, SlidingSyncStore, DelayedEventsStore, + BeeperStore, ): def __init__( self, diff --git a/synapse/storage/databases/main/account_data.py b/synapse/storage/databases/main/account_data.py index 538280137b..ada70541f2 100644 --- a/synapse/storage/databases/main/account_data.py +++ b/synapse/storage/databases/main/account_data.py @@ -20,6 +20,7 @@ # import logging +import re from typing import ( TYPE_CHECKING, Any, @@ -40,6 +41,7 @@ ) from synapse.storage.databases.main.cache import CacheInvalidationWorkerStore from synapse.storage.databases.main.push_rule import PushRulesWorkerStore +from synapse.storage.engines import PostgresEngine from synapse.storage.invite_rule import ( AllowAllInviteRulesConfig, InviteRulesConfig, @@ -57,6 +59,10 @@ logger = logging.getLogger(__name__) +# Regex pattern for detecting a bridge bot (cached here for performance) +SYNAPSE_BOT_PATTERN = re.compile(r"^@_.*_bot\:*") +HUNGRYSERV_BOT_PATTERN = re.compile(r"^@[a-z]+bot\:beeper.local") + class AccountDataWorkerStore(PushRulesWorkerStore, CacheInvalidationWorkerStore): def __init__( @@ -635,7 +641,8 @@ def process_replication_position( super().process_replication_position(stream_name, instance_name, token) async def add_account_data_to_room( - self, user_id: str, room_id: str, account_data_type: str, content: JsonDict + self, user_id: str, room_id: str, account_data_type: str, content: JsonDict, + expected_revision_id: str | None = None, ) -> int: """Add some account_data to a room for a user. @@ -644,6 +651,8 @@ async def add_account_data_to_room( room_id: The room to add a tag for. account_data_type: The type of account_data to add. content: A json object to associate with the tag. + expected_revision_id: If set, only write if the stored content's + `com.beeper.revision_id` matches (compare-and-swap). Returns: The maximum stream ID. @@ -652,9 +661,11 @@ async def add_account_data_to_room( content_json = json_encoder.encode(content) - async with self._account_data_id_gen.get_next() as next_id: - await self.db_pool.simple_upsert( - desc="add_room_account_data", + def _add_account_data_to_room_txn( + txn: LoggingTransaction, next_id: int + ) -> None: + self._upsert_account_data_txn( + txn, table="room_account_data", keyvalues={ "user_id": user_id, @@ -662,6 +673,14 @@ async def add_account_data_to_room( "account_data_type": account_data_type, }, values={"stream_id": next_id, "content": content_json}, + expected_revision_id=expected_revision_id, + ) + + async with self._account_data_id_gen.get_next() as next_id: + await self.db_pool.runInteraction( + "add_room_account_data", + _add_account_data_to_room_txn, + next_id, ) self._account_data_stream_cache.entity_has_changed(user_id, next_id) @@ -737,7 +756,8 @@ def _remove_account_data_for_room_txn( return self._account_data_id_gen.get_current_token() async def add_account_data_for_user( - self, user_id: str, account_data_type: str, content: JsonDict + self, user_id: str, account_data_type: str, content: JsonDict, + expected_revision_id: str | None = None, ) -> int: """Add some global account_data for a user. @@ -745,6 +765,8 @@ async def add_account_data_for_user( user_id: The user to add a tag for. account_data_type: The type of account_data to add. content: A json object to associate with the tag. + expected_revision_id: If set, only write if the stored content's + `com.beeper.revision_id` matches (compare-and-swap). Returns: The maximum stream ID. @@ -759,6 +781,7 @@ async def add_account_data_for_user( user_id, account_data_type, content, + expected_revision_id, ) self._account_data_stream_cache.entity_has_changed(user_id, next_id) @@ -769,6 +792,84 @@ async def add_account_data_for_user( return self._account_data_id_gen.get_current_token() + def _upsert_account_data_txn( + self, + txn: LoggingTransaction, + table: str, + keyvalues: dict[str, str], + values: dict[str, Any], + expected_revision_id: str | None, + ) -> None: + """Beeper: upsert account data, enforcing the compare-and-swap + condition when an expected revision ID is given. + + Raises a 409 SynapseError if the stored content has a (string) + `com.beeper.revision_id` that differs from the expected one. A missing + row, missing field, or non-string field matches any expected value. + """ + if expected_revision_id is None: + self.db_pool.simple_upsert_txn(txn, table, keyvalues, values) + return + + select_sql = "SELECT content FROM %s WHERE %s" % ( + table, + " AND ".join("%s = ?" % k for k in keyvalues), + ) + if isinstance(self.database_engine, PostgresEngine): + # Lock the row so concurrent CAS writes serialize. Under Synapse's + # default REPEATABLE READ isolation, a row modified by a concurrent + # transaction instead raises a serialization failure, which + # runInteraction retries with a fresh snapshot. (SQLite serializes + # writes anyway, so a plain SELECT suffices there.) + select_sql += " FOR UPDATE" + txn.execute(select_sql, list(keyvalues.values())) + row = txn.fetchone() + + if row is None: + # There is no row to lock, so a check-then-write would let two + # concurrent first writes both pass the check. Make the INSERT + # itself the atomic point instead: it only succeeds if no + # concurrent write landed first (in-flight inserts on the same + # key serialize via speculative insertion), so on success the + # no-data match genuinely held at write time. On conflict, fall + # through to compare against the winning row. + if self.db_pool.simple_upsert_txn_native_upsert( + txn, table, keyvalues, values={}, insertion_values=values + ): + return + txn.execute(select_sql, list(keyvalues.values())) + row = txn.fetchone() + if row is None: + # Under REPEATABLE READ the conflicting row was committed + # after our snapshot, so the re-read cannot see it. The + # upsert below then hits that invisible row and raises a + # serialization failure (40001), which runInteraction + # retries from scratch with a fresh snapshot that does see + # the row and compares against it. (Under READ COMMITTED + # the re-read would have found the row directly.) + self.db_pool.simple_upsert_txn(txn, table, keyvalues, values) + return + + stored_content = db_to_json(row[0]) + stored_revision_id = None + if isinstance(stored_content, dict): + rev = stored_content.get("com.beeper.revision_id") + if isinstance(rev, str): + stored_revision_id = rev + + if ( + stored_revision_id is not None + and stored_revision_id != expected_revision_id + ): + raise SynapseError( + 409, + "Account data revision ID mismatch", + Codes.EXPECTED_REVISION_ID_MISMATCH, + additional_fields={"com.beeper.current_content": stored_content}, + ) + + self.db_pool.simple_upsert_txn(txn, table, keyvalues, values) + def _add_account_data_for_user( self, txn: LoggingTransaction, @@ -776,14 +877,28 @@ def _add_account_data_for_user( user_id: str, account_data_type: str, content: JsonDict, + expected_revision_id: str | None = None, ) -> None: content_json = json_encoder.encode(content) - self.db_pool.simple_upsert_txn( + # If we're ignoring users, silently filter out any bots that may be ignored + if account_data_type == AccountDataTypes.IGNORED_USER_LIST: + ignored_users = content.get("ignored_users", {}) + if isinstance(ignored_users, dict): + content["ignored_users"] = { + u: v + for u, v in ignored_users.items() + if not ( + SYNAPSE_BOT_PATTERN.match(u) or HUNGRYSERV_BOT_PATTERN.match(u) + ) + } + + self._upsert_account_data_txn( txn, table="account_data", keyvalues={"user_id": user_id, "account_data_type": account_data_type}, values={"stream_id": next_id, "content": content_json}, + expected_revision_id=expected_revision_id, ) # Ignored users get denormalized into a separate table as an optimisation. diff --git a/synapse/storage/databases/main/beeper.py b/synapse/storage/databases/main/beeper.py new file mode 100644 index 0000000000..7c1c344644 --- /dev/null +++ b/synapse/storage/databases/main/beeper.py @@ -0,0 +1,289 @@ +# Beep beep! + +import logging +from typing import TYPE_CHECKING, cast + +from synapse.events import EventBase +from synapse.metrics.background_process_metrics import wrap_as_background_process +from synapse.storage._base import SQLBaseStore +from synapse.storage.database import ( + DatabasePool, + LoggingDatabaseConnection, + LoggingTransaction, +) +from synapse.types import RoomStreamToken +from synapse.util.duration import Duration + +if TYPE_CHECKING: + from synapse.server import HomeServer + +logger = logging.getLogger(__name__) + + +class BeeperStore(SQLBaseStore): + def __init__( + self, + database: DatabasePool, + db_conn: LoggingDatabaseConnection, + hs: "HomeServer", + ): + super().__init__(database, db_conn, hs) + + self.database = database + + self.user_notification_counts_enabled: bool = ( + hs.config.experimental.beeper_user_notification_counts_enabled + ) + + if ( + self.user_notification_counts_enabled + and hs.config.worker.run_background_tasks + ): + self.aggregate_notification_counts_loop = self.clock.looping_call( + self.beeper_aggregate_notification_counts, Duration(seconds=30) + ) + self.is_aggregating_notification_counts = False + + async def beeper_preview_event_for_room_id_and_user_id( + self, room_id: str, user_id: str, to_key: RoomStreamToken + ) -> tuple[str, int] | None: + def beeper_preview_txn(txn: LoggingTransaction) -> tuple[str, int] | None: + sql = """ + WITH latest_event AS ( + SELECT e.event_id, e.origin_server_ts + FROM events AS e + LEFT JOIN redactions as r + ON e.event_id = r.redacts + -- Look to see if this event itself is an edit, as we don't want to + -- use edits ever as the "latest event" + LEFT JOIN event_relations as is_edit + ON e.event_id = is_edit.event_id AND is_edit.relation_type = 'm.replace' + WHERE + e.stream_ordering <= ? + AND e.room_id = ? + AND is_edit.event_id IS NULL + AND r.redacts IS NULL + AND e.type IN ( + 'm.room.message', + 'm.room.encrypted', + 'm.reaction', + 'm.sticker' + ) + AND CASE + -- Only find non-redacted reactions to our own messages + WHEN (e.type = 'm.reaction') THEN ( + SELECT ? = ee.sender AND ee.event_id NOT IN ( + SELECT redacts FROM redactions WHERE redacts = ee.event_id + ) FROM events as ee + WHERE ee.event_id = ( + SELECT eer.relates_to_id FROM event_relations AS eer + WHERE eer.event_id = e.event_id + ) + ) + ELSE (true) END + ORDER BY e.stream_ordering DESC + LIMIT 1 + ), + latest_edit_for_latest_event AS ( + SELECT e.event_id, e_replacement.event_id as replacement_event_id + FROM latest_event e + -- Find any events that edit this event, as we'll want to use the new content from + -- the edit as the preview + LEFT JOIN event_relations as er + ON e.event_id = er.relates_to_id AND er.relation_type = 'm.replace' + LEFT JOIN events as e_replacement + ON er.event_id = e_replacement.event_id + ORDER BY e_replacement.origin_server_ts DESC + LIMIT 1 + ) + SELECT COALESCE(lefle.replacement_event_id, le.event_id), le.origin_server_ts + FROM latest_event le + LEFT JOIN latest_edit_for_latest_event lefle ON le.event_id = lefle.event_id + """ + + txn.execute( + sql, + ( + to_key.stream, + room_id, + user_id, + ), + ) + + return cast("tuple[str, int] | None", txn.fetchone()) + + return await self.db_pool.runInteraction( + "beeper_preview_for_room_id_and_user_id", + beeper_preview_txn, + ) + + async def beeper_cleanup_tombstoned_room(self, room_id: str) -> None: + def beeper_cleanup_tombstoned_room_txn(txn: LoggingTransaction) -> None: + self.db_pool.simple_delete_txn( + txn, table="event_push_actions", keyvalues={"room_id": room_id} + ) + self.db_pool.simple_delete_txn( + txn, table="event_push_summary", keyvalues={"room_id": room_id} + ) + + await self.db_pool.runInteraction( + "beeper_cleanup_tombstoned_room", + beeper_cleanup_tombstoned_room_txn, + ) + + def beeper_add_notification_counts_txn( + self, + txn: LoggingTransaction, + notifiable_events: list[EventBase], + ) -> None: + if not self.user_notification_counts_enabled: + return + + sql = """ + INSERT INTO beeper_user_notification_counts ( + room_id, event_stream_ordering, + user_id, thread_id, notifs, unreads, highlights + ) + SELECT ?, ?, user_id, thread_id, notif, unread, highlight + FROM event_push_actions_staging + WHERE event_id = ? + """ + + txn.execute_batch( + sql, + ( + ( + event.room_id, + event.internal_metadata.stream_ordering, + event.event_id, + ) + for event in notifiable_events + ), + ) + + def beeper_clear_notification_counts_txn( + self, + txn: LoggingTransaction, + user_id: str, + room_id: str, + stream_ordering: int, + ) -> None: + if not self.user_notification_counts_enabled: + return + + sql = """ + DELETE FROM beeper_user_notification_counts + WHERE + user_id = ? + AND room_id = ? + AND event_stream_ordering <= ? + """ + + txn.execute(sql, (user_id, room_id, stream_ordering)) + + @wrap_as_background_process("beeper_aggregate_notification_counts") + async def beeper_aggregate_notification_counts(self) -> None: + if not self.user_notification_counts_enabled: + return + + def aggregate_txn(txn: LoggingTransaction, limit: int) -> int: + sql = """ + WITH recent_rows AS ( -- Aggregate the tables, flag aggregated rows for deletion + SELECT + user_id, + room_id + FROM + beeper_user_notification_counts + WHERE + event_stream_ordering > ( + SELECT event_stream_ordering FROM beeper_user_notification_counts_stream_ordering + ) + AND event_stream_ordering < ( + -- Select highest stream ordering from events over one hour, + -- this is to avoid serialization issues with the most + -- recent events/receipts + SELECT stream_ordering FROM events + WHERE origin_server_ts < ( + (EXTRACT(EPOCH from NOW()) - 3600) * 1000 + ) + ORDER BY stream_ordering DESC + LIMIT 1 + ) + -- Oldest first, to reduce serialization issues + ORDER BY event_stream_ordering ASC + LIMIT {limit} + ) + UPDATE + beeper_user_notification_counts AS epc + SET + unreads = CASE WHEN epc.event_stream_ordering = agg.max_eso THEN agg.unreads ELSE 0 END, + notifs = CASE WHEN epc.event_stream_ordering = agg.max_eso THEN agg.notifs ELSE 0 END, + highlights = CASE WHEN epc.event_stream_ordering = agg.max_eso THEN agg.highlights ELSE 0 END, + aggregated = epc.event_stream_ordering != agg.max_eso + FROM ( + SELECT + user_id, + room_id, + SUM(unreads) AS unreads, + SUM(notifs) AS notifs, + SUM(highlights) AS highlights, + MAX(event_stream_ordering) AS max_eso + FROM + beeper_user_notification_counts + WHERE + user_id IN(SELECT user_id FROM recent_rows) + AND room_id IN(SELECT room_id FROM recent_rows) + GROUP BY + user_id, + room_id + ) AS agg + WHERE + epc.room_id = agg.room_id + AND epc.user_id = agg.user_id + RETURNING + event_stream_ordering; + """.format(limit=limit) + + txn.execute(sql) + orders = list(txn) + if not orders: + logger.info("No user counts aggregated") + return 0 + + max_stream_ordering = max(orders) + txn.execute( + """ + UPDATE beeper_user_notification_counts_stream_ordering + SET event_stream_ordering = ? + """, + (max_stream_ordering,), + ) + txn.execute("DELETE FROM beeper_user_notification_counts WHERE aggregated") + + logger.info(f"Aggregated {len(orders)} notification count rows") + + return txn.rowcount + + if self.is_aggregating_notification_counts: + return + + self.is_aggregating_notification_counts = True + limit = 1000 + + try: + logger.info("Aggregating notification counts") + + last_batch = limit + 1 + while last_batch > limit: + last_batch = await self.db_pool.runInteraction( + "beeper_aggregate_notification_counts", + aggregate_txn, + limit=limit, + ) + await self.clock.sleep(Duration(seconds=1)) + + except self.database.engine.module.OperationalError: + logger.exception("Failed to aggregate notifications") + + finally: + self.is_aggregating_notification_counts = False diff --git a/synapse/storage/databases/main/client_ips.py b/synapse/storage/databases/main/client_ips.py index f9e36ea8be..38bfa345e1 100644 --- a/synapse/storage/databases/main/client_ips.py +++ b/synapse/storage/databases/main/client_ips.py @@ -20,6 +20,7 @@ # import logging +from os import environ from typing import ( TYPE_CHECKING, Mapping, @@ -51,8 +52,10 @@ # Number of msec of granularity to store the user IP 'last seen' time. Smaller # times give more inserts into the database even for readonly API hits -# 120 seconds == 2 minutes -LAST_SEEN_GRANULARITY = 120 * 1000 +# 120 seconds == 2 minutes, Beep: updated to 1h +LAST_SEEN_GRANULARITY = 3600 * 1000 + +DISABLE_CLIENT_IP_STORAGE = environ.get("SYNAPSE_DISABLE_CLIENT_IP_STORAGE") == "true" @attr.s(slots=True, frozen=True, auto_attribs=True) @@ -710,14 +713,16 @@ def _update_client_ips_batch_txn( devices_keys.append((user_id, device_id)) devices_values.append((user_agent, last_seen, ip)) - self.db_pool.simple_upsert_many_txn( - txn, - table="user_ips", - key_names=("user_id", "access_token", "ip"), - key_values=user_ips_keys, - value_names=("user_agent", "device_id", "last_seen"), - value_values=user_ips_values, - ) + # Beep: only store user_ips if not disabled + if not DISABLE_CLIENT_IP_STORAGE: + self.db_pool.simple_upsert_many_txn( + txn, + table="user_ips", + key_names=("user_id", "access_token", "ip"), + key_values=user_ips_keys, + value_names=("user_agent", "device_id", "last_seen"), + value_values=user_ips_values, + ) if devices_values: self.db_pool.simple_update_many_txn( diff --git a/synapse/storage/databases/main/event_push_actions.py b/synapse/storage/databases/main/event_push_actions.py index 7cecbcb65c..9035adc2ac 100644 --- a/synapse/storage/databases/main/event_push_actions.py +++ b/synapse/storage/databases/main/event_push_actions.py @@ -1154,7 +1154,7 @@ async def add_push_actions_to_staging( self, event_id: str, user_id_actions: dict[str, Collection[Mapping | str]], - count_as_unread: bool, + count_as_unread_by_user: dict[str, bool], thread_id: str, ) -> None: """Add the push actions for the event to the push action staging area. @@ -1163,7 +1163,8 @@ async def add_push_actions_to_staging( event_id user_id_actions: A mapping of user_id to list of push actions, where an action can either be a string or dict. - count_as_unread: Whether this event should increment unread counts. + count_as_unread_by_user: Whether this event should increment unread + counts for each user. thread_id: The thread this event is parent of, if applicable. """ if not user_id_actions: @@ -1182,7 +1183,7 @@ def _gen_entry( _serialize_action(actions, bool(is_highlight)), # actions column notif, # notif column is_highlight, # highlight column - int(count_as_unread), # unread column + int(count_as_unread_by_user.get(user_id, 0)), # unread column thread_id, # thread_id column self.clock.time_msec(), # inserted_ts column ) diff --git a/synapse/storage/databases/main/events.py b/synapse/storage/databases/main/events.py index ef2110292d..83f5c3c2ae 100644 --- a/synapse/storage/databases/main/events.py +++ b/synapse/storage/databases/main/events.py @@ -3987,6 +3987,8 @@ def _set_push_actions_for_event_and_users_txn( ], ) + self.store.beeper_add_notification_counts_txn(txn, notifiable_events) + # Now we delete the staging area for *all* events that were being # persisted. txn.execute_batch( diff --git a/synapse/storage/databases/main/purge_events.py b/synapse/storage/databases/main/purge_events.py index f564d3322a..64f11f175f 100644 --- a/synapse/storage/databases/main/purge_events.py +++ b/synapse/storage/databases/main/purge_events.py @@ -518,7 +518,6 @@ def _purge_room_txn(self, txn: LoggingTransaction, room_id: str) -> None: (room_id,), ) - if isinstance(self.database_engine, PostgresEngine): # Disable statement timeouts for this transaction; purging rooms can # take a while! txn.execute("SET LOCAL statement_timeout = 0") diff --git a/synapse/storage/databases/main/receipts.py b/synapse/storage/databases/main/receipts.py index 3720d25c1d..b234dc8646 100644 --- a/synapse/storage/databases/main/receipts.py +++ b/synapse/storage/databases/main/receipts.py @@ -1009,6 +1009,11 @@ def _insert_linearized_receipt_txn( where_clause=where_clause, ) + if self.hs.is_mine_id(user_id): + self.beeper_clear_notification_counts_txn( # type: ignore[attr-defined] + txn, user_id, room_id, stream_ordering + ) + return rx_ts def _graph_to_linear( diff --git a/synapse/storage/schema/main/delta/73/99_beeper_user_notification_counts.sql b/synapse/storage/schema/main/delta/73/99_beeper_user_notification_counts.sql new file mode 100644 index 0000000000..674cc4d3ec --- /dev/null +++ b/synapse/storage/schema/main/delta/73/99_beeper_user_notification_counts.sql @@ -0,0 +1,19 @@ +CREATE TABLE beeper_user_notification_counts ( + user_id TEXT, + room_id TEXT, + thread_id TEXT, + event_stream_ordering BIGINT, + notifs BIGINT, + unreads BIGINT, + highlights BIGINT, + aggregated BOOLEAN, + UNIQUE (user_id, room_id, thread_id, event_stream_ordering) +); + +CREATE TABLE beeper_user_notification_counts_stream_ordering ( + lock CHAR(1) NOT NULL DEFAULT 'X' UNIQUE, -- Makes sure this table only has one row. + event_stream_ordering BIGINT NOT NULL, + CHECK (lock='X') +); + +INSERT INTO beeper_user_notification_counts_stream_ordering (event_stream_ordering) VALUES (0); diff --git a/synapse/util/task_scheduler.py b/synapse/util/task_scheduler.py index aef3eb681c..a9c779b9c6 100644 --- a/synapse/util/task_scheduler.py +++ b/synapse/util/task_scheduler.py @@ -110,6 +110,8 @@ class TaskScheduler: # Report a running task's status and usage every so often. OCCASIONAL_REPORT_INTERVAL = Duration(minutes=5) + SLEEP_AFTER_TASK_S = 1 + def __init__(self, hs: "HomeServer"): self.hs = hs # nb must be called this for @wrap_as_background_process self.server_name = hs.hostname @@ -360,7 +362,7 @@ async def on_cancel_task(self, id: str) -> None: if id in self._running_tasks: deferred = self._running_tasks[id] deferred.cancel() - self._running_tasks.pop(id) + self._running_tasks.pop(id, None) await self.update_task(id, status=TaskStatus.CANCELLED) def on_new_task(self, task_id: str) -> None: @@ -499,14 +501,23 @@ async def wrapper() -> None: result=result, error=error, ) - self._running_tasks.pop(task.id) + + # Beep: throttle between scheduled tasks without hiding the + # terminal status from callers. + try: + if TaskScheduler.SLEEP_AFTER_TASK_S > 0: + await self._clock.sleep( + Duration(seconds=TaskScheduler.SLEEP_AFTER_TASK_S) + ) + finally: + self._running_tasks.pop(task.id, None) + occasional_status_call.stop() current_time = self._clock.time() usage = log_context.get_resource_usage() TaskScheduler._log_task_usage( status.value, task, usage, current_time - start_time ) - occasional_status_call.stop() # Try launch a new task since we've finished with this one. self._clock.call_later( diff --git a/tests/events/test_utils.py b/tests/events/test_utils.py index c5f473186e..83d308dcbd 100644 --- a/tests/events/test_utils.py +++ b/tests/events/test_utils.py @@ -805,6 +805,14 @@ def test_default_serialize_config_excludes_admin_metadata(self) -> None: ).include_admin_metadata ) + def test_stream_ordering_is_included(self) -> None: + event = make_test_event({"unsigned": {"com.beeper.hs.order": 999}}) + event.internal_metadata.stream_ordering = 1234 + + serialized = self.serialize(event, fields=None) + + self.assertEqual(serialized["unsigned"]["com.beeper.hs.order"], 1234) + def test_event_flagged_for_admins(self) -> None: # Default behaviour should be *not* to include it self.assertEqual( diff --git a/tests/push/test_bulk_push_rule_evaluator.py b/tests/push/test_bulk_push_rule_evaluator.py index f6be933871..c742bff169 100644 --- a/tests/push/test_bulk_push_rule_evaluator.py +++ b/tests/push/test_bulk_push_rule_evaluator.py @@ -545,7 +545,7 @@ def test_suppress_edits(self) -> None: ) # An edit which is a mention will cause a notification. - self.assertTrue( + self.assertFalse( # Beeper: changed from true per our base rule changes self._create_and_process( bulk_evaluator, { diff --git a/tests/push/test_http.py b/tests/push/test_http.py index bf6d2ce084..b014fad7be 100644 --- a/tests/push/test_http.py +++ b/tests/push/test_http.py @@ -420,8 +420,8 @@ def test_sends_high_priority_for_one_to_one_only(self) -> None: self.push_attempts[1][1], "http://example.com/_matrix/push/v1/notify" ) - # check that this is low-priority - self.assertEqual(self.push_attempts[1][2]["notification"]["prio"], "low") + # Beeper: all notifications are high priority + self.assertEqual(self.push_attempts[1][2]["notification"]["prio"], "high") def test_sends_high_priority_for_user_mention(self) -> None: """ @@ -507,8 +507,8 @@ def test_sends_high_priority_for_user_mention(self) -> None: self.push_attempts[1][1], "http://example.com/_matrix/push/v1/notify" ) - # check that this is low-priority - self.assertEqual(self.push_attempts[1][2]["notification"]["prio"], "low") + # Beeper: all notifications are high priority + self.assertEqual(self.push_attempts[1][2]["notification"]["prio"], "high") def test_sends_high_priority_for_atroom(self) -> None: """ @@ -603,8 +603,8 @@ def test_sends_high_priority_for_atroom(self) -> None: self.push_attempts[1][1], "http://example.com/_matrix/push/v1/notify" ) - # check that this is low-priority - self.assertEqual(self.push_attempts[1][2]["notification"]["prio"], "low") + # Beeper: all notifications are high priority + self.assertEqual(self.push_attempts[1][2]["notification"]["prio"], "high") def test_push_unread_count_group_by_room(self) -> None: """ diff --git a/tests/push/test_push_rule_evaluator.py b/tests/push/test_push_rule_evaluator.py index 2e389710b9..0bf7ad28ed 100644 --- a/tests/push/test_push_rule_evaluator.py +++ b/tests/push/test_push_rule_evaluator.py @@ -33,7 +33,7 @@ from synapse.rest.client import login, register, room from synapse.server import HomeServer from synapse.storage.databases.main.appservice import _make_exclusive_regex -from synapse.synapse_rust.push import PushRuleEvaluator +from synapse.synapse_rust.push import FilteredPushRules, PushRuleEvaluator, PushRules from synapse.types import JsonDict, JsonMapping, UserID from synapse.util.clock import Clock from synapse.util.frozenutils import freeze @@ -1079,3 +1079,149 @@ def test_delayed_message(self) -> None: # user2 should not be notified about it, because they can't see it. self.assertEqual(self.get_notif_count(self.user_id2), 0) + + +class PushRuleEvaluatorBaseRulesTestCase(unittest.TestCase): + def test_reactions(self) -> None: + message_event = make_test_event( + { + "event_id": "$event_id", + "room_id": "!room_id:beeper.com", + "content": { + "body": "Looks like Nick is way ahead of me on this one", + "msgtype": "m.text", + }, + "sender": "@brad:beeper.com", + "type": "m.room.message", + }, + RoomVersions.V1, + ) + + reaction_event = make_test_event( + { + "event_id": "$reaction_id", + "room_id": "!room_id:beeper.com", + "content": { + "m.relates_to": { + "event_id": "$event_id", + "key": "\U0001f44d", + "rel_type": "m.annotation", + } + }, + "sender": "@nick:beeper.com", + "type": "m.reaction", + }, + RoomVersions.V1, + ) + + dm_evaluator = PushRuleEvaluator( + _flatten_dict(reaction_event), + False, + 2, + 0, + {}, + {"m.annotation": _flatten_dict(message_event)}, + True, + reaction_event.room_version.msc3931_push_features, + True, + True, + False, + ) + + # Reaction to Brad's message, should be an action for Brad + actions = dm_evaluator.run( + FilteredPushRules( + PushRules([]), {}, True, True, True, True, True, True, True + ), + "@brad:beeper.com", + "Brad", + False, + ) + self.assertTrue("notify" in actions) + + # Reaction to Brad's message, should not be an action for Nick + actions = dm_evaluator.run( + FilteredPushRules( + PushRules([]), {}, True, True, True, True, True, True, True + ), + "@nick:beeper.com", + "Nick", + False, + ) + self.assertEqual(actions, []) + + large_room_evaluator = PushRuleEvaluator( + _flatten_dict(reaction_event), + False, + 30, + 0, + {}, + {"m.annotation": _flatten_dict(message_event)}, + True, + reaction_event.room_version.msc3931_push_features, + True, + True, + False, + ) + + # Large rooms should never have emoji reaction notifications + actions = large_room_evaluator.run( + FilteredPushRules( + PushRules([]), {}, True, True, True, True, True, True, True + ), + "@brad:beeper.com", + "Brad", + False, + ) + self.assertEqual(actions, []) + actions = large_room_evaluator.run( + FilteredPushRules( + PushRules([]), {}, True, True, True, True, True, True, True + ), + "@nick:beeper.com", + "Nick", + False, + ) + self.assertEqual(actions, []) + + def test_supress_auto_accept_invite(self) -> None: + event = make_test_event( + { + "event_id": "$event_id", + "room_id": "!wFyjEwanOaElpGOaLW:beeper.com", + "content": { + "displayname": "Brad Murray", + "fi.mau.will_auto_accept": True, + "is_direct": True, + "membership": "invite", + }, + "sender": "@_brad_imessagecloud_83372:beeper.com", + "state_key": "@brad:beeper.com", + "type": "m.room.member", + }, + RoomVersions.V1, + ) + + evaluator = PushRuleEvaluator( + _flatten_dict(event), + False, + 0, + 0, + {}, + {}, + True, + event.room_version.msc3931_push_features, + True, + True, + False, + ) + + actions = evaluator.run( + FilteredPushRules( + PushRules([]), {}, True, True, True, True, True, True, True + ), + "@brad:beeper.com", + "Brad Murray", + False, + ) + self.assertEqual(actions, []) diff --git a/tests/replication/storage/test_events.py b/tests/replication/storage/test_events.py index d7e6dfca83..bfb206e65c 100644 --- a/tests/replication/storage/test_events.py +++ b/tests/replication/storage/test_events.py @@ -291,7 +291,7 @@ def build_event( self.master_store.add_push_actions_to_staging( event.event_id, dict(push_actions), - False, + {user_id: False for user_id, _ in push_actions}, "main", ) ) diff --git a/tests/rest/admin/test_scheduled_tasks.py b/tests/rest/admin/test_scheduled_tasks.py index 388570df0b..fc3699a014 100644 --- a/tests/rest/admin/test_scheduled_tasks.py +++ b/tests/rest/admin/test_scheduled_tasks.py @@ -73,6 +73,8 @@ async def _failed_test_task( "failed_test_task", resource_id="failed_task" ) ) + self.reactor.advance(self._task_scheduler.SLEEP_AFTER_TASK_S) + self.reactor.advance(0.2) def check_scheduled_tasks_response(self, scheduled_tasks: Mapping) -> list: result = [] diff --git a/tests/rest/client/test_account_data.py b/tests/rest/client/test_account_data.py index be6d7af2fc..aff873d870 100644 --- a/tests/rest/client/test_account_data.py +++ b/tests/rest/client/test_account_data.py @@ -20,6 +20,7 @@ # from unittest.mock import AsyncMock +from synapse.api.constants import ReceiptTypes from synapse.rest import admin from synapse.rest.client import account_data, login, room @@ -79,3 +80,135 @@ def test_on_account_data_updated_callback(self) -> None: mocked_callback.assert_called_with( user_id, room_id, account_data_type, account_data_content ) + + def test_beeper_inbox_state_endpoint(self) -> None: + store = self.hs.get_datastores().main + + user_id = self.register_user("user", "password") + tok = self.login("user", "password") + + room_id = self.helper.create_room_as(user_id, tok=tok) + channel = self.make_request( + "PUT", + f"/_matrix/client/unstable/com.beeper.inbox/user/{user_id}/rooms/{room_id}/inbox_state", + {}, + access_token=tok, + ) + + self.assertEqual(channel.code, 200, channel.result) + self.assertIsNone( + self.get_success( + store.get_account_data_for_room_and_type( + user_id, room_id, "com.beeper.inbox.done" + ) + ) + ) + self.assertIsNone( + self.get_success( + store.get_account_data_for_room_and_type( + user_id, room_id, "m.marked_unread" + ) + ) + ) + + before_ts = self.clock.time_msec() + channel = self.make_request( + "PUT", + f"/_matrix/client/unstable/com.beeper.inbox/user/{user_id}/rooms/{room_id}/inbox_state", + {"done": {"at_delta": 1000 * 60 * 5}, "marked_unread": True}, + access_token=tok, + ) + after_ts = self.clock.time_msec() + + self.assertEqual(channel.code, 200, channel.result) + + done = self.get_success( + store.get_account_data_for_room_and_type( + user_id, room_id, "com.beeper.inbox.done" + ) + ) + assert done is not None + self.assertGreaterEqual(done["updated_ts"], before_ts) + self.assertLessEqual(done["updated_ts"], after_ts) + self.assertEqual(done["at_ts"], done["updated_ts"] + (1000 * 60 * 5)) + + marked_unread = self.get_success( + store.get_account_data_for_room_and_type( + user_id, room_id, "m.marked_unread" + ) + ) + assert marked_unread is not None + self.assertEqual(marked_unread["unread"], True) + self.assertEqual(marked_unread["ts"], done["updated_ts"]) + + def test_beeper_inbox_state_endpoint_can_clear_unread(self) -> None: + store = self.hs.get_datastores().main + + user_id = self.register_user("user", "password") + tok = self.login("user", "password") + + room_id = self.helper.create_room_as(user_id, tok=tok) + before_ts = self.clock.time_msec() + channel = self.make_request( + "PUT", + f"/_matrix/client/unstable/com.beeper.inbox/user/{user_id}/rooms/{room_id}/inbox_state", + {"marked_unread": False}, + access_token=tok, + ) + after_ts = self.clock.time_msec() + + self.assertEqual(channel.code, 200, channel.result) + + self.assertEqual(channel.code, 200, channel.result) + self.assertIsNone( + self.get_success( + store.get_account_data_for_room_and_type( + user_id, room_id, "com.beeper.inbox.done" + ) + ) + ) + + marked_unread = self.get_success( + store.get_account_data_for_room_and_type( + user_id, room_id, "m.marked_unread" + ) + ) + assert marked_unread is not None + self.assertEqual(marked_unread["unread"], False) + self.assertGreaterEqual(marked_unread["ts"], before_ts) + self.assertLessEqual(marked_unread["ts"], after_ts) + + def test_beeper_inbox_state_endpoint_can_set_read_marker(self) -> None: + store = self.hs.get_datastores().main + + user_id = self.register_user("user", "password") + tok = self.login("user", "password") + + room_id = self.helper.create_room_as(user_id, tok=tok) + + res = self.helper.send(room_id, "hello", tok=tok) + + existing_read_marker = self.get_success( + store.get_account_data_for_room_and_type( + user_id, room_id, ReceiptTypes.FULLY_READ + ) + ) + + channel = self.make_request( + "PUT", + f"/_matrix/client/unstable/com.beeper.inbox/user/{user_id}/rooms/{room_id}/inbox_state", + { + "read_markers": { + ReceiptTypes.FULLY_READ: res["event_id"], + }, + }, + access_token=tok, + ) + self.assertEqual(channel.code, 200) + + new_read_marker = self.get_success( + store.get_account_data_for_room_and_type( + user_id, room_id, ReceiptTypes.FULLY_READ + ) + ) + self.assertNotEqual(existing_read_marker, new_read_marker) diff --git a/tests/rest/client/test_relations.py b/tests/rest/client/test_relations.py index 2d8ba77a77..ea9d4b1b1b 100644 --- a/tests/rest/client/test_relations.py +++ b/tests/rest/client/test_relations.py @@ -21,6 +21,7 @@ import urllib.parse from typing import Any, Callable +from unittest import skip as stdlib_skip from unittest.mock import AsyncMock, patch from twisted.internet.testing import MemoryReactor @@ -406,6 +407,7 @@ def _assert_edit_bundle( m_replace_dict, ) + @stdlib_skip("Beeper disables bundled aggregations") def test_edit(self) -> None: """Test that a simple edit works.""" orig_body = {"body": "Hi!", "msgtype": "m.text"} @@ -494,6 +496,7 @@ def test_edit(self) -> None: edit_event_content, ) + @stdlib_skip("Beeper disables bundled aggregations") def test_multi_edit(self) -> None: """Test that multiple edits, including attempts by people who shouldn't be allowed, are correctly handled. @@ -544,6 +547,7 @@ def test_multi_edit(self) -> None: channel.json_body["event"], edit_event_id, edit_event_content ) + @stdlib_skip("Beeper disables bundled aggregations") def test_edit_reply(self) -> None: """Test that editing a reply works.""" @@ -609,6 +613,7 @@ def test_edit_reply(self) -> None: result_event_dict, edit_event_id, edit_event_content ) + @stdlib_skip("Beeper disables bundled aggregations") def test_edit_edit(self) -> None: """Test that an edit cannot be edited.""" orig_body = {"body": "Hi!", "msgtype": "m.text"} @@ -1072,6 +1077,7 @@ def test_recursive_relations_with_filter(self) -> None: self.assertEqual(event_ids, [annotation_1]) +@stdlib_skip("Beeper disables bundled aggregations") class BundledAggregationsTestCase(BaseRelationsTestCase): """ See RelationsTestCase.test_edit for a similar test for edits. @@ -1478,6 +1484,7 @@ def test_bundled_aggregations_with_filter(self) -> None: ) +@stdlib_skip("Beeper disables bundled aggregations") class RelationIgnoredUserTestCase(BaseRelationsTestCase): """Relations sent from an ignored user should be ignored.""" @@ -1574,6 +1581,7 @@ def test_thread(self) -> None: ) +@stdlib_skip("Beeper disables bundled aggregations") class RelationRedactionTestCase(BaseRelationsTestCase): """ Test the behaviour of relations when the parent or child event is redacted. @@ -1802,6 +1810,7 @@ def _get_threads(self, body: JsonDict) -> list[tuple[str, str]]: for ev in body["chunk"] ] + @stdlib_skip("Beeper disables bundled aggregations") def test_threads(self) -> None: """Create threads and ensure the ordering is due to their latest event.""" # Create 2 threads. diff --git a/tests/rest/client/test_sync.py b/tests/rest/client/test_sync.py index bf00c0f331..775f16d8c6 100644 --- a/tests/rest/client/test_sync.py +++ b/tests/rest/client/test_sync.py @@ -35,7 +35,16 @@ RelationTypes, ) from synapse.events import EventBase -from synapse.rest.client import devices, knock, login, read_marker, receipts, room, sync +from synapse.rest.client import ( + account_data, + devices, + knock, + login, + read_marker, + receipts, + room, + sync, +) from synapse.server import HomeServer from synapse.types import JsonDict, RoomStreamToken, StreamKeyType, StreamToken from synapse.util.clock import Clock @@ -470,6 +479,7 @@ def test_create_event_present_in_knock_state(self) -> None: class UnreadMessagesTestCase(unittest.HomeserverTestCase): servlets = [ + account_data.register_servlets, synapse.rest.admin.register_servlets, login.register_servlets, read_marker.register_servlets, @@ -571,6 +581,9 @@ def test_unread_counts(self) -> None: # Check that the unread counter is back to 0. self._check_unread_count(0) + # Beeper: we don't count name as unread, so send this to increase the counter + self.helper.send_event(self.room_id, EventTypes.Encrypted, {}, tok=self.tok2) + # Check that room name changes increase the unread counter. self.helper.send_state( self.room_id, @@ -580,6 +593,9 @@ def test_unread_counts(self) -> None: ) self._check_unread_count(1) + # Beeper: we don't count topic as unread, so send this to increase the counter + self.helper.send_event(self.room_id, EventTypes.Encrypted, {}, tok=self.tok2) + # Check that room topic changes increase the unread counter. self.helper.send_state( self.room_id, @@ -593,6 +609,10 @@ def test_unread_counts(self) -> None: self.helper.send_event(self.room_id, EventTypes.Encrypted, {}, tok=self.tok2) self._check_unread_count(3) + # Beeper: fake event to bump event count, we don't count custom events + # as unread currently. + self.helper.send_event(self.room_id, EventTypes.Encrypted, {}, tok=self.tok2) + # Check that custom events with a body increase the unread counter. result = self.helper.send_event( self.room_id, @@ -626,7 +646,7 @@ def test_unread_counts(self) -> None: content={"body": "hello", "msgtype": "m.notice"}, tok=self.tok2, ) - self._check_unread_count(4) + self._check_unread_count(5) # Beep: notices count as unread # Check that tombstone events changes increase the unread counter. res1 = self.helper.send_state( @@ -657,6 +677,28 @@ def test_unread_counts(self) -> None: self.assertEqual(channel.code, 200, channel.json_body) self._check_unread_count(0) + def test_beeper_inbox_state_can_update_unread_count(self) -> None: + # increase unread count + self.helper.join(room=self.room_id, user=self.user2, tok=self.tok2) + res = self.helper.send(self.room_id, "hello", tok=self.tok2) + self._check_unread_count(1) + + # Beeper: inbox_state should be able to send read receipts + res = self.helper.send(self.room_id, "hello", tok=self.tok2) + + channel = self.make_request( + "PUT", + f"/_matrix/client/unstable/com.beeper.inbox/user/{self.user_id}/rooms/{self.room_id}/inbox_state", + { + "read_markers": { + ReceiptTypes.READ: res["event_id"], + }, + }, + access_token=self.tok, + ) + self.assertEqual(channel.code, 200, channel.json_body) + self._check_unread_count(0) + # We test for all three receipt types that influence notification counts @parameterized.expand( [ @@ -2138,3 +2180,348 @@ def test_membership_change_sent_when_cache_claims_it_was( response = self._sync(use_state_after, t0, timeline_limit=1) self.assertNotIn(kick["event_id"], self._timeline_ids(response)) self.assertIn(kick["event_id"], self._state_ids(response, use_state_after)) + + +class BeeperRoomPreviewTestCase(unittest.HomeserverTestCase): + servlets = [ + synapse.rest.admin.register_servlets, + login.register_servlets, + read_marker.register_servlets, + room.register_servlets, + sync.register_servlets, + receipts.register_servlets, + ] + + def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: + self.url = "/sync?beeper_previews=true&since=%s" + self.next_batches = {} + + # Register the first user (used to check the unread counts). + self.user_id = self.register_user("kermit", "monkey") + self.tok = self.login("kermit", "monkey") + self.next_batches[self.tok] = "s0" + + # Create the room we'll check unread counts for. + self.room_id = self.helper.create_room_as(self.user_id, tok=self.tok) + self.room_id_2 = self.helper.create_room_as(self.user_id, tok=self.tok) + self.room_id_3 = self.helper.create_room_as(self.user_id, tok=self.tok) + self.room_id_4 = self.helper.create_room_as(self.user_id, tok=self.tok) + + # Register the second user (used to send events to the room). + self.user2 = self.register_user("kermit2", "monkey") + self.tok2 = self.login("kermit2", "monkey") + self.next_batches[self.tok2] = "s0" + + # Change the power levels of the room so that the second user can send state + # events. + self.helper.send_state( + self.room_id, + EventTypes.PowerLevels, + { + "users": {self.user2: 100}, + "users_default": 0, + "events": { + "m.room.name": 50, + "m.room.power_levels": 100, + "m.room.history_visibility": 100, + "m.room.canonical_alias": 50, + "m.room.avatar": 50, + "m.room.tombstone": 100, + "m.room.server_acl": 100, + "m.room.encryption": 100, + }, + "events_default": 0, + "state_default": 50, + "ban": 50, + "kick": 50, + "redact": 50, + "invite": 0, + }, + tok=self.tok, + ) + + def _check_preview_event_ids(self, auth_token: str, expected: dict) -> None: + """Checks the populated preview value against the expected value provided""" + + channel = self.make_request( + "GET", + self.url % self.next_batches[auth_token], + access_token=auth_token, + ) + + self.assertEqual(channel.code, 200, channel.json_body) + + for room_id, expected_entry in expected.items(): + room_entry = ( + channel.json_body.get("rooms", {}).get("join", {}).get(room_id, {}) + ) + + preview = room_entry.get("com.beeper.inbox.preview") + if preview: + preview_id = preview.get("event_id") + self.assertEqual( + preview_id, + expected_entry, + room_entry, + ) + else: + self.assertIsNone(expected_entry, room_entry) + + # Store the next batch for the next request. + self.next_batches[auth_token] = channel.json_body["next_batch"] + + def _redact_event( + self, + access_token: str, + room_id: str, + event_id: str, + expect_code: int = 200, + with_relations: list[str] | None = None, + ) -> JsonDict: + """Helper function to send a redaction event. + + Returns the json body. + """ + path = "/_matrix/client/r0/rooms/%s/redact/%s" % (room_id, event_id) + + request_content = {} + if with_relations: + request_content["org.matrix.msc3912.with_relations"] = with_relations + + channel = self.make_request( + "POST", path, request_content, access_token=access_token + ) + self.assertEqual(channel.code, expect_code) + return channel.json_body + + def test_room_previews(self) -> None: + """Tests that /sync returns all room previews on first sync.""" + + # Multiple events in rooms for first sync. + self.helper.join(room=self.room_id, user=self.user2, tok=self.tok2) + self.helper.join(room=self.room_id_2, user=self.user2, tok=self.tok2) + self.helper.join(room=self.room_id_3, user=self.user2, tok=self.tok2) + self.helper.join(room=self.room_id_4, user=self.user2, tok=self.tok2) + + send_body = self.helper.send(self.room_id, "hello", tok=self.tok2) + send_body2 = self.helper.send(self.room_id_2, "hello 2", tok=self.tok2) + send_body3 = self.helper.send(self.room_id_3, "hello 3", tok=self.tok2) + send_body4 = self.helper.send(self.room_id_4, "hello 4", tok=self.tok2) + + # Should have previews for all rooms on first sync. + self._check_preview_event_ids( + auth_token=self.tok, + expected={ + self.room_id: send_body["event_id"], + self.room_id_2: send_body2["event_id"], + self.room_id_3: send_body3["event_id"], + self.room_id_4: send_body4["event_id"], + }, + ) + + # Subsequent - update preview for only room 2" + send_body5 = self.helper.send(self.room_id_2, "Sup!", tok=self.tok2) + + self._check_preview_event_ids( + auth_token=self.tok, expected={self.room_id_2: send_body5["event_id"]} + ) + + def test_room_preview(self) -> None: + """Tests that /sync returns a room preview with the latest message for room.""" + + # One user says hello. + # Check that a message we send returns a preview in the room (i.e. have multiple clients?) + send_body = self.helper.send(self.room_id, "hello", tok=self.tok) + self._check_preview_event_ids( + auth_token=self.tok, expected={self.room_id: send_body["event_id"]} + ) + + # Join new user. Should not show updated preview. + self.helper.join(room=self.room_id, user=self.user2, tok=self.tok2) + self._check_preview_event_ids( + auth_token=self.tok, expected={self.room_id: send_body["event_id"]} + ) + + # Second user says hello + # Check that the new user sending a message updates our preview + send_2_body = self.helper.send(self.room_id, "hello again!", tok=self.tok2) + self._check_preview_event_ids(self.tok, {self.room_id: send_2_body["event_id"]}) + + # Encrypted messages 1 + # Beeper: ensure encrypted messages are treated the same. + enc_1_body = self.helper.send_event( + self.room_id, EventTypes.Encrypted, {}, tok=self.tok2 + ) + self._check_preview_event_ids( + auth_token=self.tok, expected={self.room_id: enc_1_body["event_id"]} + ) + + # Encrypted messages 2 + enc_2_body = self.helper.send_event( + self.room_id, EventTypes.Encrypted, {}, tok=self.tok2 + ) + self._check_preview_event_ids( + auth_token=self.tok, expected={self.room_id: enc_2_body["event_id"]} + ) + + # Redact encrypted message 2 + self._redact_event(self.tok2, self.room_id, enc_2_body["event_id"]) + self._check_preview_event_ids( + auth_token=self.tok, expected={self.room_id: enc_1_body["event_id"]} + ) + + # User 2 react to user 1 message + # Someone else reacted to my message, update preview. + reaction_1 = self.helper.send_event( + room_id=self.room_id, + type=EventTypes.Reaction, + content={ + "m.relates_to": { + "rel_type": RelationTypes.ANNOTATION, + "event_id": send_body["event_id"], + "key": "👍", + } + }, + tok=self.tok2, + ) + self._check_preview_event_ids( + auth_token=self.tok, expected={self.room_id: reaction_1["event_id"]} + ) + + # User 1 react to User 2 message. + # Not a reaction to my message, don't update preview. + reaction_2 = self.helper.send_event( + room_id=self.room_id, + type=EventTypes.Reaction, + content={ + "m.relates_to": { + "rel_type": RelationTypes.ANNOTATION, + "event_id": send_2_body["event_id"], + "key": "👍", + } + }, + tok=self.tok, + ) + self._check_preview_event_ids( + auth_token=self.tok, expected={self.room_id: reaction_1["event_id"]} + ) + self._check_preview_event_ids( + auth_token=self.tok2, expected={self.room_id: reaction_2["event_id"]} + ) + + # Redact user 2 message with reactions. + # Remove redactions as well as reactions from user 2's preview. + self._redact_event(self.tok2, self.room_id, send_2_body["event_id"]) + + self._check_preview_event_ids( + auth_token=self.tok, expected={self.room_id: reaction_1["event_id"]} + ) + self._check_preview_event_ids( + auth_token=self.tok2, expected={self.room_id: enc_1_body["event_id"]} + ) + + def test_room_preview_edits(self) -> None: + """Tests that /sync returns a room preview with the latest message for room.""" + + # One user says hello. + # Check that a message we send returns a preview in the room (i.e. have multiple clients?) + send_body = self.helper.send(self.room_id, "hello", tok=self.tok) + self._check_preview_event_ids( + auth_token=self.tok, expected={self.room_id: send_body["event_id"]} + ) + + # Join new user. Should not show updated preview. + self.helper.join(room=self.room_id, user=self.user2, tok=self.tok2) + self._check_preview_event_ids( + auth_token=self.tok, expected={self.room_id: send_body["event_id"]} + ) + + # Second user says hello + # Check that the new user sending a message updates our preview + send_2_body = self.helper.send(self.room_id, "hello again!", tok=self.tok2) + self._check_preview_event_ids(self.tok, {self.room_id: send_2_body["event_id"]}) + + # First user edits their old message + # Check that this doesn't alter the preview + self.helper.send_event( + room_id=self.room_id, + type=EventTypes.Message, + content={ + "body": "hello edit", + "msgtype": "m.text", + "m.relates_to": { + "rel_type": RelationTypes.REPLACE, + "event_id": send_body["event_id"], + }, + }, + tok=self.tok, + ) + self._check_preview_event_ids(self.tok, {self.room_id: send_2_body["event_id"]}) + + # Now second user edits their (currently preview) message + # Check that this does become the preview + send_3_body = self.helper.send_event( + room_id=self.room_id, + type=EventTypes.Message, + content={ + "body": "hello edit", + "msgtype": "m.text", + "m.relates_to": { + "rel_type": RelationTypes.REPLACE, + "event_id": send_2_body["event_id"], + }, + }, + tok=self.tok2, + ) + self._check_preview_event_ids(self.tok, {self.room_id: send_3_body["event_id"]}) + + # Now second user edits their (currently preview) message again + # Check that this does become the preview, over the previous edit + send_4_body = self.helper.send_event( + room_id=self.room_id, + type=EventTypes.Message, + content={ + "body": "hello edit 2", + "msgtype": "m.text", + "m.relates_to": { + "rel_type": RelationTypes.REPLACE, + "event_id": send_2_body["event_id"], + }, + }, + tok=self.tok2, + ) + self._check_preview_event_ids(self.tok, {self.room_id: send_4_body["event_id"]}) + + # Finally, first user sends a message and this should become the preview + send_5_body = self.helper.send(self.room_id, "hello", tok=self.tok) + self._check_preview_event_ids( + auth_token=self.tok, expected={self.room_id: send_5_body["event_id"]} + ) + + def test_room_preview_no_change(self) -> None: + """Tests that /sync only includes previews when we have new events.""" + + self.helper.join(room=self.room_id, user=self.user_id, tok=self.tok) + + send_body = self.helper.send(self.room_id, "hello", tok=self.tok) + + # Should have preview on first sync + self._check_preview_event_ids( + auth_token=self.tok, + expected={self.room_id: send_body["event_id"]}, + ) + + # Should have no preview on second sync (no timeline changes) + self._check_preview_event_ids( + auth_token=self.tok, + expected={self.room_id: None}, + ) + + # Send a join event, this isn't previewed but will be in the timeline + self.helper.join(room=self.room_id, user=self.user2, tok=self.tok2) + + # Should have preview because we have timeline, but preview is unchanged + self._check_preview_event_ids( + auth_token=self.tok, + expected={self.room_id: send_body["event_id"]}, + ) diff --git a/tests/rest/client/test_upgrade_room.py b/tests/rest/client/test_upgrade_room.py index cc4073cafc..395a354c21 100644 --- a/tests/rest/client/test_upgrade_room.py +++ b/tests/rest/client/test_upgrade_room.py @@ -26,7 +26,7 @@ from synapse.api.room_versions import RoomVersions from synapse.config.server import DEFAULT_ROOM_VERSION from synapse.rest import admin -from synapse.rest.client import login, room, room_upgrade_rest_servlet +from synapse.rest.client import login, notifications, room, room_upgrade_rest_servlet from synapse.server import HomeServer from synapse.util.clock import Clock @@ -40,6 +40,7 @@ class UpgradeRoomTest(unittest.HomeserverTestCase): login.register_servlets, room.register_servlets, room_upgrade_rest_servlet.register_servlets, + notifications.register_servlets, ] def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: @@ -485,3 +486,37 @@ def test_creator_removed_from_powerlevels_v12(self) -> None: # This is a regression test where previously Synapse would accidentally # mutate the old power levels event. self.assertEqual(old_power_level_event.content["users"][self.creator], 100) + + def test_upgrade_clears_push_actions(self) -> None: + """ + Beeper specific test: ensure that when upgrading a room any notification/unread counts + in the old room are removed. + """ + self.helper.send_event( + self.room_id, + "m.room.message", + content={"body": "hi", "msgtype": "text"}, + tok=self.other_token, + ) + + # Check we have a notification pre-upgrade + channel = self.make_request( + "GET", + "/notifications", + access_token=self.creator_token, + ) + self.assertEqual(channel.code, 200, channel.result) + self.assertEqual(len(channel.json_body["notifications"]), 1, channel.json_body) + + channel = self._upgrade_room() + self.assertEqual(200, channel.code, channel.result) + self.assertIn("replacement_room", channel.json_body) + + # Check we have no notification post-upgrade + channel = self.make_request( + "GET", + "/notifications", + access_token=self.creator_token, + ) + self.assertEqual(channel.code, 200, channel.result) + self.assertEqual(len(channel.json_body["notifications"]), 0, channel.json_body) diff --git a/tests/storage/test_account_data.py b/tests/storage/test_account_data.py index c91aad097d..72549406ae 100644 --- a/tests/storage/test_account_data.py +++ b/tests/storage/test_account_data.py @@ -108,6 +108,22 @@ def test_ignoring_self_fails(self) -> None: self.assertEqual(f.code, 400) self.assertEqual(f.errcode, Codes.INVALID_PARAM) + def test_ignoring_bot_users(self) -> None: + self._update_ignore_list("@other:test", "@another:remote") + self.assert_ignored(self.user, {"@other:test", "@another:remote"}) + + self._update_ignore_list("@other:test", "@another:remote", "@_other_bot:test") + self.assert_ignored(self.user, {"@other:test", "@another:remote"}) + + self._update_ignore_list("@iamnotabot:beeper.com") + self.assert_ignored(self.user, {"@iamnotabot:beeper.com"}) + + self._update_ignore_list("@_other_bot:beeper.com") + self.assert_ignored(self.user, set()) + + self._update_ignore_list("@whatsappbot:beeper.local") + self.assert_ignored(self.user, set()) + def test_caching(self) -> None: """Ensure that caching works properly between different users.""" # The first user ignores a user. diff --git a/tests/storage/test_client_ips.py b/tests/storage/test_client_ips.py index fd138335f0..69c15b5e9a 100644 --- a/tests/storage/test_client_ips.py +++ b/tests/storage/test_client_ips.py @@ -100,7 +100,7 @@ def test_insert_new_client_ip_none_device_id(self) -> None: user_id, "access_token", "ip", "user_agent", None ) ) - self.reactor.advance(200) + self.reactor.advance(3600) self.pump(0) result = cast( @@ -153,7 +153,7 @@ def test_insert_new_client_ip_none_device_id(self) -> None: ) # Only one result, has been upserted. self.assertEqual( - result, [("access_token", "ip", "user_agent", None, 12345878000)] + result, [("access_token", "ip", "user_agent", None, 12349278000)] ) @parameterized.expand([(False,), (True,)]) diff --git a/tests/util/test_task_scheduler.py b/tests/util/test_task_scheduler.py index bdd4f21924..55c1652eff 100644 --- a/tests/util/test_task_scheduler.py +++ b/tests/util/test_task_scheduler.py @@ -142,12 +142,14 @@ def get_tasks_of_status(status: TaskStatus) -> list[ScheduledTask]: self.assertEqual(len(scheduled_tasks), 1) # The scheduled task should start 0.1s after the first of the active tasks - # finishes - self.reactor.advance(0.1) + # finishes its post-task throttle. + self.reactor.advance(TaskScheduler.SLEEP_AFTER_TASK_S) + self.reactor.advance(0.2) self.assertEqual(len(get_tasks_of_status(TaskStatus.ACTIVE)), 1) - # ... and should finally complete after another second - self.reactor.advance(1) + # ... and should finally complete after it runs. + self.reactor.advance(TaskScheduler.SLEEP_AFTER_TASK_S) + self.reactor.advance(0.2) prev_scheduled_task = self.get_success( self.task_scheduler.get_task(scheduled_tasks[0].id) ) @@ -201,12 +203,14 @@ def get_tasks_of_status(status: TaskStatus) -> list[ScheduledTask]: self.assertEqual(len(scheduled_tasks), 1) # The scheduled task should start 0.1s after the first of the active tasks - # finishes - self.reactor.advance(0.1) + # finishes its post-task throttle. + self.reactor.advance(TaskScheduler.SLEEP_AFTER_TASK_S) + self.reactor.advance(0.2) self.assertEqual(len(get_tasks_of_status(TaskStatus.ACTIVE)), 1) - # ... and should finally complete after another second - self.reactor.advance(1) + # ... and should finally complete after it runs. + self.reactor.advance(TaskScheduler.SLEEP_AFTER_TASK_S) + self.reactor.advance(0.2) prev_scheduled_task = self.get_success( self.task_scheduler.get_task(scheduled_tasks[0].id) ) @@ -270,7 +274,8 @@ def _test_cancel_task(self, task_id: str) -> None: assert task.result and "counter" in task.result current_counter = int(task.result["counter"]) - self.reactor.advance(1) + self.reactor.advance(TaskScheduler.SLEEP_AFTER_TASK_S) + self.reactor.advance(0.2) task = self.get_success(self.task_scheduler.get_task(task_id)) assert task is not None