Compare commits

..

117 Commits

Author SHA1 Message Date
Zomatree
59af536e29 refactor: seperate out disconnect logic 2025-10-12 02:02:36 +01:00
Zomatree
5c7eec68d0 chore: add debug logging to voice ingress 2025-10-12 01:12:08 +01:00
Zomatree
94d7320e70 fix(ci): add voice-ingress to compose 2025-10-11 23:46:13 +01:00
Zomatree
af78ac0586 chore(ci): move to stoatchat 2025-10-11 23:22:41 +01:00
Zomatree
a42ddcbbfa Merge branch 'main' into feat/livekit 2025-10-11 23:09:07 +01:00
Zomatree
d65c1a1ab3 fix(ci): publish images under stoatchat and remove docker hub 2025-10-11 23:01:35 +01:00
Zomatree
530264f747 chore: add error logging to internal errors 2025-10-11 22:22:11 +01:00
Zomatree
ae1d4dd264 fix: dont presume channel has a node 2025-10-09 01:32:17 +01:00
izzy
4789dd4847 fix: include voice-ingress in build 2025-09-30 14:55:31 -05:00
izzy
f91d3bbfdd merge: branch 'main' into feat/livekit 2025-09-30 14:52:24 -05:00
izzy
4fb99e3bd0 ci: also include gifbox src 2025-09-23 12:12:33 -05:00
izzy
f0a83abcfa ci: add missing src copies for Docker build 2025-09-23 12:09:19 -05:00
izzy
6f1c715b8c chore: bump version to 0.8.9 2025-09-23 12:02:35 -05:00
Zomatree
5b50532d15 feat: track join time 2025-09-20 05:26:58 +01:00
Zomatree
938480ccec Signed-off-by: Angelo Kontaxis <me@zomatree.live> 2025-09-20 01:58:57 +01:00
Zomatree
38dd4d1079 fix: swap to using reqwest for query building 2025-09-19 02:50:00 +01:00
Zomatree
154204742d feat: add ratelimits to gifbox 2025-09-19 02:50:00 +01:00
Zomatree
db55998546 docs: document revolt-coalesced 2025-09-19 02:50:00 +01:00
Zomatree
5885e067a6 feat: trending and categories routes 2025-09-19 02:50:00 +01:00
Zomatree
a92152d86d fix: use our own result types instead of tenors types
add user auth
2025-09-19 02:50:00 +01:00
Zomatree
b5cd5e30ef feat: require auth for search 2025-09-19 02:50:00 +01:00
Zomatree
b0c977b324 feat: initial work on tenor gif searching 2025-09-19 02:50:00 +01:00
Zomatree
bfe4018e43 fix: apple music to use original url instead of metadata url
Apple Music returns the url without the query parameter in the opengraph meta tags, causing the regex to not find the track id, meaning any embed links generated would only link to the album.
2025-09-19 02:41:53 +01:00
Zomatree
cc7a7962a8 fix: use trust_cloudflare config value instead of env var 2025-09-18 21:22:31 +01:00
Zomatree
fb4011084d refactor: move ratelimits to a generic system for all web servers 2025-09-18 21:22:31 +01:00
Zomatree
3a3415915f fix: relax settings name regex 2025-09-17 15:41:54 +01:00
Zomatree
964884a5de chore: add policy changes 2025-09-17 15:41:54 +01:00
Zomatree
db57706794 feat: ready payload field customisation 2025-09-17 15:41:54 +01:00
Paul Makles
a186732ca5 merge: pull request #428 from ElfFlu/bugfix/16byte-attachment-gain 2025-09-12 12:48:28 +01:00
Zomatree
f422e8e917 Merge remote-tracking branch 'origin' into feat/livekit 2025-09-07 05:39:51 +01:00
Zomatree
1632e3dacf feat: allow users to specify who is notified when starting a call 2025-09-07 05:01:14 +01:00
Tom
e635bc23ec fix: don't remove timeouts when a member leaves a server (#409) 2025-09-06 18:41:23 -07:00
Zomatree
cc9a68b9c5 chore: cleanup 2025-09-07 02:27:50 +01:00
IAmTomahawkx
77a23239ea feat: Send push notifications for dm call start/end 2025-09-06 15:14:48 -07:00
Zomatree
3cb7da95e3 chore: add missing traits implementations to PermissionValue 2025-09-05 15:22:55 +01:00
Zomatree
dea0f675dd fix: implement Serialize to ClientMessage 2025-09-05 13:07:52 +01:00
Zomatree
480f210ce8 feat: include member payload in ServerMemberJoin event 2025-08-24 06:32:38 +01:00
Zomatree
caa8607468 feat: include groups and dms in fetch mutuals 2025-08-24 03:40:10 +01:00
ElfFlu
32e6600272 fix: remove authentication tag bytes from attachment download
Signed-off-by: ElfFlu <228066338+ElfFlu@users.noreply.github.com>
2025-08-23 19:05:32 +02:00
Zomatree
5335124306 fix: add license to revolt-parser 2025-08-20 18:55:52 +01:00
Zomatree
1561481eb4 feat: make message lexer use unowned string 2025-08-20 18:55:39 +01:00
Zomatree
d03e6be2eb fix: update migration 2025-08-19 18:56:17 +01:00
Zomatree
0a4d77ffd3 fix: remove VoiceChannel channel type
fix: calculate user overwrites correctly
fix: dont include personal info in livekit user metadata
fix: revoke video permissons on denied publish
fix: add video to default permissions
2025-08-19 17:37:09 +01:00
IAmTomahawkx
8f9015a6ff fix: apply uname fix to january and autumn 2025-08-18 16:59:08 -07:00
Zomatree
c6f121f585 Merge remote-tracking branch 'origin/main' into feat/livekit 2025-08-18 07:44:52 +01:00
Zomatree
96ddeb79d2 fix: dont set max participants in livekit 2025-08-18 07:28:09 +01:00
Zomatree
a36b138978 fix: move call system message to daemon,
check max participants when creating a token to avoid giving tokens but erroring when attempting to join,
check if the channel actually supports voice
2025-08-18 04:58:06 +01:00
Zomatree
860c47fbad feat: - track call length
- move voice and video limits to config
- seperate VoiceInformation into model and db model
- fix build scripts
2025-08-12 01:48:57 +01:00
Zomatree
a770d5822a chore: cleanup code 2025-08-04 03:20:08 +01:00
Zomatree
10c9682bba feat: ability to force disconnect existing voice connection 2025-08-04 03:19:53 +01:00
Zomatree
eb48975428 chore: bump livekit dep 2025-07-01 20:22:53 +01:00
Zomatree
28ed85a779 Merge branch 'main' into feat/livekit 2025-07-01 20:22:34 +01:00
Zomatree
95ca735b7b fix: get_voice_state has incorrect unique_key and allowed_sources should be lowercase 2025-06-08 02:33:35 +01:00
Zomatree
901ac56154 fix: remove node from channel voice state 2025-06-05 17:26:37 +01:00
Zomatree
3dbf3bb7a0 feat: private livekit nodes 2025-06-05 14:14:19 +01:00
Zomatree
45df5724b3 Merge remote-tracking branch 'origin/main' into livekit 2025-06-02 02:20:04 +01:00
Zomatree
ea1c165d80 Merge remote-tracking branch 'origin/main' into livekit 2025-06-02 02:11:51 +01:00
Zomatree
48e0b48ea9 fix: update branch from main 2025-05-01 16:58:35 +01:00
Zomatree
61038ddcf1 Merge remote-tracking branch 'origin/main' into livekit 2025-05-01 16:42:13 +01:00
Zomatree
b96f07e615 fix: update branch from main 2025-04-29 15:52:03 +01:00
Zomatree
d830fb0911 Merge branch 'main' into livekit 2025-04-29 15:47:16 +01:00
Zomatree
cc9ff9a632 fix: add back missing early adopter badge 2025-04-29 15:40:41 +01:00
Zomatree
02f18705f4 fix: dont leak invisible presence to others 2025-04-29 15:40:10 +01:00
Zomatree
a841df3b36 fix: show full branch name on github webhook messages 2025-04-29 15:40:10 +01:00
Tom
2a36f5912d feat: Add Mass Mentions to the backend (#394)
* feat: create base of push daemon

Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com>

* Add outbound senders

* Make web_push send to rabbit instead (temp stuff)

* feat: stability and friend requests

* make vapid fr stuff not suck

* swap naming of queue

* move pushd into daemons folder

* fix cargo file for move into daemons folder

* feat: probably working fcm push notifs

* comment out fcm webpush stuff since the config keys dont exist

* fix fcm, name queues according to their prod status and configure routing keys

* add pushd to docker

* mix: Remove old code, add stuff to pushd

* fix: lockfile

* feat: update rocket to 5.0.1

* fix: fix queues and ack bugs

* Move rabbit messsage processing into ack queue

* chore: update readme

* chore: optimizations for ack database hits

* pushd flowchart

* misc: update flowchart

* exit dependancy hell

* add rocket_impl flag to authifier

* make the tests file of delta actually compile

* fix: don't silence every push message

* fix: don't silence all messages

* add debug logging for sending data to rabbit from message events

* validate mentions at a server membership level

* put back that import that was actually important

* minor fix to lockfile

* update delta authifier

* feat: proper permissions for push notifications

* add unit test for mention sanitization

* remove local file dependancy on authifier

* update ports to proper defaults

* fixTM the node bindings

* Theoretically configure docker releases for pushd

Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com>

* declare exchange in pushd and delta

* fix createbuckets script

Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com>

* fix: reference db implementation

Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com>

* fix: remove finally redundant code

Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com>

* fix: changes

Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com>

* fix: other changes

Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com>

* fix: make channel name return result

Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com>

* Add role mention parsing

* feat: update to mongo 3.1, add member generator.

* integrate mass mentions into pushd

* patch redis-rs with updated versions

* feat: chunk role mentions

* move permission bits to 37/38 to avoid livekit conflict

* change role mention format to <%id>

* fix the lockfile from merge

* fix: PR change requests

* feat: add tests

* fix: i am a dumbass

* fix: tests, again

---------

Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com>
2025-04-29 15:40:09 +01:00
izzy
a110a087be ci: downgrade lockfile 2025-04-29 15:39:12 +01:00
izzy
21b9164ee7 chore: bump version to 0.8.4
chore: enable webhooks in test builds
2025-04-29 15:39:12 +01:00
ShaksterNano
2f3d1f0c7e feat: add option to send message with missing replies
Signed-off-by: ShaksterNano <54268387+shaksternano@users.noreply.github.com>
2025-04-29 15:39:12 +01:00
Builderb
62204fe50b feat: allow bots to manage emojis (#407)
* Remove bot check on emoji create

Signed-off-by: Builderb <builderbgamer@gmail.com>

* Remove bot check on emoji delete

Signed-off-by: Builderb <builderbgamer@gmail.com>

---------

Signed-off-by: Builderb <builderbgamer@gmail.com>
2025-04-29 15:39:12 +01:00
Bradlee Barnes
773b890012 fix: remove SavedMessages and Group due to given permissions
Signed-off-by: Bradlee Barnes <69256931+StupidRepo@users.noreply.github.com>
2025-04-29 15:39:12 +01:00
Bradlee Barnes
9c3b2fb364 fix: allow message pinning and unpinning for DM-like channels
Signed-off-by: Bradlee Barnes <69256931+StupidRepo@users.noreply.github.com>
2025-04-29 15:39:11 +01:00
Nils Ponsard
5fe19d9815 feat: allow to set an icon when creating a group
Signed-off-by: Nils Ponsard <nilsponsard@gmail.com>
2025-04-29 15:39:11 +01:00
phazeschift
c1c5b0fd29 test: add delete server channel test
Signed-off-by: phazeschift <51278042+phazeschift@users.noreply.github.com>
2025-04-29 15:39:11 +01:00
phazeschift
6737100b4b test: add utilities for common setup
Signed-off-by: phazeschift <51278042+phazeschift@users.noreply.github.com>
2025-04-29 15:39:11 +01:00
phazeschift
6f0e80eef9 test: Add tests for invite/fetch
Signed-off-by: phazeschift <51278042+phazeschift@users.noreply.github.com>
2025-04-29 15:39:11 +01:00
IAmTomahawkx
bfc6f50ee5 chore: ignore python venv and dev script for user generation
Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com>
2025-04-29 15:39:11 +01:00
Paul Makles
7abbad700c fix: include production default value for config 2025-04-29 15:39:11 +01:00
Paul Makles
a671ba6308 fix: don't bump the lockfile version 2025-04-29 15:39:11 +01:00
Paul Makles
1d385ef2bc chore: bump version to 0.8.3 2025-04-29 15:39:11 +01:00
Paul Makles
6558e9f7c4 feat: add company information to email footers 2025-04-29 15:39:10 +01:00
Paul Makles
92f4b44cb3 chore: clarify cargo deny / tomls 2025-04-29 15:39:10 +01:00
Paul Makles
370442bf4e feat: add crond container definitions 2025-04-29 15:39:10 +01:00
Paul Makles
d54c5e80fa chore: fix version references 2025-04-29 15:39:10 +01:00
Paul Makles
a42746c55b feat: file deletion implementation 2025-04-29 15:39:10 +01:00
Paul Makles
9a98c53aff feat: init crond crate 2025-04-29 15:38:13 +01:00
Paul Makles
4ea08a189d fix: match new error type for status code 2025-04-29 15:37:42 +01:00
Paul Makles
ae15a0b4c7 fix: don't allow users to time themselves out
closes #376
2025-04-29 15:37:42 +01:00
Paul Makles
9e5fbec9b0 chore: disable LTO because it likely overloads GitHub worker 2025-04-29 15:36:56 +01:00
Paul Makles
0b02c84f04 chore: bump version to 0.8.2 2025-04-29 15:36:56 +01:00
Paul Makles
8bfe3ec6b0 fix: ensure ratelimiter adjusts for version prefix 2025-04-29 15:36:56 +01:00
Paul Makles
a7345a7b90 chore: mount API at /0.8/ as well 2025-04-29 15:36:56 +01:00
Paul Makles
12854d0f1b fix: change permission check for fetching channel webhooks 2025-04-29 15:36:56 +01:00
Paul Makles
9a2a7901e7 fix: ensure limit is always at least 1 when sent to database 2025-04-29 15:36:56 +01:00
Paul Makles
e8cb0bb88a docs: add some notes on overrides; fix defaults 2025-04-29 15:36:55 +01:00
Paul Makles
343b90b9af refactor: add error messages for Rabbit issues 2025-04-29 15:36:55 +01:00
Paul Makles
a4246f39e4 chore: add pushd to debug image script 2025-04-29 15:36:55 +01:00
Zomatree
38380db6e7 refactor: Remove VoiceChannel 2025-04-29 15:05:18 +01:00
Zomatree
091aa4705b fix: use existing node if someone is already in the voice channel 2025-03-25 20:37:06 +00:00
Zomatree
8b5e085232 feat: being moved between voice channels 2025-03-01 22:09:40 +00:00
Zomatree
9c2cc8736c fix: respond with the url when joining a voice channel 2025-02-17 02:11:32 +00:00
Zomatree
951128032b fix: expose list of nodes 2025-02-17 01:48:34 +00:00
Zomatree
367ac887f4 feat: support multiple voice nodes 2025-02-17 01:39:00 +00:00
Zomatree
060b4c43f4 feat: make Channel::server return a reference 2025-02-07 21:35:51 +00:00
Zomatree
2593a4272b chore: switch to ids for parameter 2025-02-02 13:46:45 +00:00
Zomatree
742eb3bb7d fix: remove local testing values from config 2025-01-28 16:06:58 +00:00
Zomatree
41a6de67cd fix: remove locked rocket version 2025-01-28 15:59:58 +00:00
Zomatree
c8ae34ba73 feat: implement missing permission syncs 2025-01-28 15:55:21 +00:00
Zomatree
c55b5bf75f fix: include voice states in servercreate event
feat: call started system message in dms
2025-01-27 20:46:29 +00:00
Zomatree
9de87a01ab Merge remote-tracking branch 'origin/main' into livekit 2025-01-26 20:36:40 +00:00
Zomatree
2f7294705e feat(livekit): push unfinished code 2024-10-13 20:23:00 +01:00
Zomatree
120ca449b8 feat(livekit): more permission handling 2024-07-23 04:07:34 +01:00
Zomatree
ffbc899792 split voice ops into its own library 2024-05-13 14:09:08 +01:00
Zomatree
675c7d6c0d fix redis key inconsistancy 2024-04-24 02:29:31 +01:00
Zomatree
67569f932e more stuff 2024-04-24 02:23:52 +01:00
Zomatree
7a4421089c track current state 2024-04-11 01:07:03 +01:00
Zomatree
ae87bc0b67 move voice states to global 2024-04-09 22:36:48 +01:00
Zomatree
ad0dcad497 initial livekit support
fix up code

undo changes to compose file

add back .env.example
2024-04-09 21:21:52 +01:00
232 changed files with 7060 additions and 2714 deletions

View File

@@ -49,13 +49,6 @@ jobs:
uses: docker/setup-buildx-action@v2
# Authenticate with Docker Hub and GHCR
- name: Login to DockerHub
uses: docker/login-action@v2
with:
registry: docker.io
username: ${{ secrets.DOCKERHUB_USERNAME }}
password: ${{ secrets.DOCKERHUB_TOKEN }}
- name: Login to Github Container Registry
uses: docker/login-action@v2
with:
@@ -72,14 +65,13 @@ jobs:
platforms: linux/amd64,linux/arm64
tags: ghcr.io/${{ github.repository_owner }}/base:latest
# revoltchat/server
# stoatchat/api
- name: Docker meta
id: meta-delta
uses: docker/metadata-action@v4
with:
images: |
docker.io/revoltchat/server
ghcr.io/revoltchat/server
ghcr.io/stoatchat/api
- name: Publish
uses: docker/build-push-action@v4
with:
@@ -92,14 +84,13 @@ jobs:
BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:latest
labels: ${{ steps.meta-delta.outputs.labels }}
# revoltchat/bonfire
# stoatchat/events
- name: Docker meta
id: meta-bonfire
uses: docker/metadata-action@v4
with:
images: |
docker.io/revoltchat/bonfire
ghcr.io/revoltchat/bonfire
ghcr.io/stoatchat/events
- name: Publish
uses: docker/build-push-action@v4
with:
@@ -112,14 +103,13 @@ jobs:
BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:latest
labels: ${{ steps.meta-bonfire.outputs.labels }}
# revoltchat/autumn
# stoatchat/file-server
- name: Docker meta
id: meta-autumn
uses: docker/metadata-action@v4
with:
images: |
docker.io/revoltchat/autumn
ghcr.io/revoltchat/autumn
ghcr.io/stoatchat/file-server
- name: Publish
uses: docker/build-push-action@v4
with:
@@ -132,14 +122,13 @@ jobs:
BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:latest
labels: ${{ steps.meta-autumn.outputs.labels }}
# revoltchat/january
# stoatchat/proxy
- name: Docker meta
id: meta-january
uses: docker/metadata-action@v4
with:
images: |
docker.io/revoltchat/january
ghcr.io/revoltchat/january
ghcr.io/stoatchat/proxy
- name: Publish
uses: docker/build-push-action@v4
with:
@@ -152,14 +141,32 @@ jobs:
BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:latest
labels: ${{ steps.meta-january.outputs.labels }}
# revoltchat/crond
# stoatchat/gifbox
- name: Docker meta
id: meta-gifbox
uses: docker/metadata-action@v4
with:
images: |
ghcr.io/stoatchat/gifbox
- name: Publish
uses: docker/build-push-action@v4
with:
context: .
push: true
platforms: linux/amd64,linux/arm64
file: crates/services/gifbox/Dockerfile
tags: ${{ steps.meta-gifbox.outputs.tags }}
build-args: |
BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:latest
labels: ${{ steps.meta-gifbox.outputs.labels }}
# stoatchat/crond
- name: Docker meta
id: meta-crond
uses: docker/metadata-action@v4
with:
images: |
docker.io/revoltchat/crond
ghcr.io/revoltchat/crond
ghcr.io/stoatchat/crond
- name: Publish
uses: docker/build-push-action@v4
with:
@@ -172,14 +179,13 @@ jobs:
BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:latest
labels: ${{ steps.meta-crond.outputs.labels }}
# revoltchat/pushd
# stoatchat/pushd
- name: Docker meta
id: meta-pushd
uses: docker/metadata-action@v4
with:
images: |
docker.io/revoltchat/pushd
ghcr.io/revoltchat/pushd
ghcr.io/stoatchat/pushd
- name: Publish
uses: docker/build-push-action@v4
with:
@@ -191,3 +197,22 @@ jobs:
build-args: |
BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:latest
labels: ${{ steps.meta-pushd.outputs.labels }}
# stoatchat/voice-ingress
- name: Docker meta
id: meta-voice-ingress
uses: docker/metadata-action@v4
with:
images: |
ghcr.io/stoatchat/voice-ingress
- name: Publish
uses: docker/build-push-action@v4
with:
context: .
push: true
platforms: linux/amd64,linux/arm64
file: crates/daemons/voice-ingress/Dockerfile
tags: ${{ steps.meta-voice-ingress.outputs.tags }}
build-args: |
BASE_IMAGE=ghcr.io/${{ github.repository_owner }}/base:latest
labels: ${{ steps.meta-voice-ingress.outputs.labels }}

View File

@@ -73,7 +73,7 @@ jobs:
if: github.event_name != 'pull_request' && github.ref_name == 'main'
uses: actions/checkout@v3
with:
repository: revoltchat/api
repository: stoatchat/api
path: api
token: ${{ secrets.PAT }}

View File

@@ -14,7 +14,7 @@ jobs:
run: |
gh api graphql -f query='
query {
organization(login: "revoltchat"){
organization(login: "stoatchat"){
projectV2(number: 3) {
id
fields(first:20) {

View File

@@ -14,7 +14,7 @@ jobs:
run: |
gh api graphql -f query='
query {
organization(login: "revoltchat"){
organization(login: "stoatchat"){
projectV2(number: 5) {
id
fields(first:20) {

4
.gitignore vendored
View File

@@ -11,4 +11,6 @@ venv/
.vercel
.DS_Store
.idea
livekit.yml
.idea
start

2043
Cargo.lock generated

File diff suppressed because it is too large Load Diff

View File

@@ -27,10 +27,14 @@ COPY crates/core/parser/Cargo.toml ./crates/core/parser/
COPY crates/core/permissions/Cargo.toml ./crates/core/permissions/
COPY crates/core/presence/Cargo.toml ./crates/core/presence/
COPY crates/core/result/Cargo.toml ./crates/core/result/
COPY crates/core/coalesced/Cargo.toml ./crates/core/coalesced/
COPY crates/core/ratelimits/Cargo.toml ./crates/core/ratelimits/
COPY crates/services/autumn/Cargo.toml ./crates/services/autumn/
COPY crates/services/january/Cargo.toml ./crates/services/january/
COPY crates/services/gifbox/Cargo.toml ./crates/services/gifbox/
COPY crates/daemons/crond/Cargo.toml ./crates/daemons/crond/
COPY crates/daemons/pushd/Cargo.toml ./crates/daemons/pushd/
COPY crates/daemons/voice-ingress/Cargo.toml ./crates/daemons/voice-ingress/
RUN sh /tmp/build-image-layer.sh deps
# Build all apps

View File

@@ -23,10 +23,14 @@ COPY crates/core/parser/Cargo.toml ./crates/core/parser/
COPY crates/core/permissions/Cargo.toml ./crates/core/permissions/
COPY crates/core/presence/Cargo.toml ./crates/core/presence/
COPY crates/core/result/Cargo.toml ./crates/core/result/
COPY crates/core/coalesced/Cargo.toml ./crates/core/coalesced/
COPY crates/core/ratelimits/Cargo.toml ./crates/core/ratelimits/
COPY crates/services/autumn/Cargo.toml ./crates/services/autumn/
COPY crates/services/january/Cargo.toml ./crates/services/january/
COPY crates/services/gifbox/Cargo.toml ./crates/services/gifbox/
COPY crates/daemons/crond/Cargo.toml ./crates/daemons/crond/
COPY crates/daemons/pushd/Cargo.toml ./crates/daemons/pushd/
COPY crates/daemons/voice-ingress/Cargo.toml ./crates/daemons/voice-ingress/
RUN sh /tmp/build-image-layer.sh deps
# Build all apps

View File

@@ -21,9 +21,11 @@ The services and libraries that power the Revolt service.<br/>
| `core/permissions` | [crates/core/permissions](crates/core/permissions) | Core: Permission Logic | ![Crates.io Version](https://img.shields.io/crates/v/revolt-permissions) ![Crates.io Version](https://img.shields.io/crates/msrv/revolt-permissions) ![Crates.io Version](https://img.shields.io/crates/size/revolt-permissions) ![Crates.io License](https://img.shields.io/crates/l/revolt-permissions) |
| `core/presence` | [crates/core/presence](crates/core/presence) | Core: User Presence | ![Crates.io Version](https://img.shields.io/crates/v/revolt-presence) ![Crates.io Version](https://img.shields.io/crates/msrv/revolt-presence) ![Crates.io Version](https://img.shields.io/crates/size/revolt-presence) ![Crates.io License](https://img.shields.io/crates/l/revolt-presence) |
| `core/result` | [crates/core/result](crates/core/result) | Core: Result and Error types | ![Crates.io Version](https://img.shields.io/crates/v/revolt-result) ![Crates.io Version](https://img.shields.io/crates/msrv/revolt-result) ![Crates.io Version](https://img.shields.io/crates/size/revolt-result) ![Crates.io License](https://img.shields.io/crates/l/revolt-result) |
| `core/coalesced` | [crates/core/coalesced](crates/core/coalesced) | Core: Coalescion service | ![Crates.io Version](https://img.shields.io/crates/v/revolt-coalesced) ![Crates.io Version](https://img.shields.io/crates/msrv/revolt-coalesced) ![Crates.io Version](https://img.shields.io/crates/size/revolt-coalesced) ![Crates.io License](https://img.shields.io/crates/l/revolt-coalesced) |
| `delta` | [crates/delta](crates/delta) | REST API server | ![License](https://img.shields.io/badge/license-AGPL--3.0--or--later-blue) |
| `bonfire` | [crates/bonfire](crates/bonfire) | WebSocket events server | ![License](https://img.shields.io/badge/license-AGPL--3.0--or--later-blue) |
| `services/january` | [crates/services/january](crates/services/january) | Proxy server | ![License](https://img.shields.io/badge/license-AGPL--3.0--or--later-blue) |
| `services/gifbox` | [crates/services/gifbox](crates/services/gifbox) | Tenor proxy server | ![License](https://img.shields.io/badge/license-AGPL--3.0--or--later-blue) |
| `services/autumn` | [crates/services/autumn](crates/services/autumn) | File server | ![License](https://img.shields.io/badge/license-AGPL--3.0--or--later-blue) |
| `daemons/crond` | [crates/daemons/crond](crates/daemons/crond) | Timed data clean up daemon server | ![License](https://img.shields.io/badge/license-AGPL--3.0--or--later-blue) |
| `daemons/pushd` | [crates/daemons/pushd](crates/daemons/pushd) | Push notification daemon server | ![License](https://img.shields.io/badge/license-AGPL--3.0--or--later-blue) |
@@ -66,6 +68,7 @@ As a heads-up, the development environment uses the following ports:
| `crates/bonfire` | 14703 |
| `crates/services/autumn` | 14704 |
| `crates/services/january` | 14705 |
| `crates/services/gifbox` | 14706 |
Now you can clone and build the project:
@@ -141,6 +144,8 @@ cargo run --bin revolt-bonfire
cargo run --bin revolt-autumn
# run the proxy server
cargo run --bin revolt-january
# run the tenor proxy
cargo run --bin revolt-gifbox
# run the push daemon (not usually needed in regular development)
cargo run --bin revolt-pushd

View File

@@ -26,6 +26,11 @@ january = "http://local.revolt.chat:14705"
voso_legacy = ""
voso_legacy_ws = ""
# Public urls for livekit nodes
# each entry here should have a corresponding entry under `api.livekit.nodes`
[hosts.livekit]
worldwide = "ws://local.revolt.chat:14706"
[api]
[api.smtp]
@@ -40,6 +45,18 @@ port = 14025
use_tls = false
use_starttls = false
[api.livekit]
# Config for livekit nodes
# Make sure to change the secret when deploying
# The key and secret should match the values livekit is using
[api.livekit.nodes.worldwide]
url = "http://livekit"
lat = 0.0
lon = 0.0
key = "worldwide_key"
secret = "ZjCofRlfm6GGtjlifmNpCDkcQbEIIVC0"
[files.s3]
# S3 protocol endpoint
endpoint = "http://127.0.0.1:14009"

View File

@@ -26,7 +26,3 @@ disallowed-methods = [
# Prefer to use Object::delete(&self)
"revolt_database::models::bots::ops::AbstractBots::delete_bot",
]
disallowed-types = [
"rocket::serde::json::Json",
]

View File

@@ -40,7 +40,7 @@ services:
# Rabbit
rabbit:
image: rabbitmq:3-management
image: rabbitmq:4-management
environment:
RABBITMQ_DEFAULT_USER: rabbituser
RABBITMQ_DEFAULT_PASS: rabbitpass

View File

@@ -1,6 +1,6 @@
[package]
name = "revolt-bonfire"
version = "0.8.8"
version = "0.8.9"
license = "AGPL-3.0-or-later"
edition = "2021"
@@ -19,6 +19,7 @@ async-channel = "2.3.1"
# parsing
querystring = "1.1.0"
regex = "1.11.1"
# serde
bincode = "1.3.3"
@@ -37,11 +38,11 @@ async-std = { version = "1.8.0", features = [
# core
authifier = { version = "1.0.15" }
revolt-result = { path = "../core/result", features = ["sentry"] }
revolt-result = { path = "../core/result" }
revolt-models = { path = "../core/models" }
revolt-config = { path = "../core/config" }
revolt-database = { path = "../core/database" }
revolt-permissions = { version = "0.8.8", path = "../core/permissions" }
revolt-permissions = { version = "0.8.9", path = "../core/permissions" }
revolt-presence = { path = "../core/presence", features = ["redis-is-patched"] }
# redis

View File

@@ -1,5 +1,5 @@
# Build Stage
FROM ghcr.io/revoltchat/base:latest AS builder
FROM ghcr.io/stoatchat/base:latest AS builder
FROM debian:12 AS debian
# Bundle Stage

View File

@@ -1,9 +1,15 @@
use async_tungstenite::tungstenite::{handshake, Message};
use futures::channel::oneshot::Sender;
use once_cell::sync::Lazy;
use regex::Regex;
use revolt_database::events::client::ReadyPayloadFields;
use revolt_result::{create_error, Result, ToRevoltError};
use revolt_result::{create_error, Result};
use serde::{Deserialize, Serialize};
/// matches either a single word ie "users" or a key and value ie "settings[notifications]"
static READY_PAYLOAD_FIELD_REGEX: Lazy<Regex> =
Lazy::new(|| Regex::new(r#"^(\w+)(?:\[(\S+)\])?$"#).unwrap());
/// Enumeration of supported protocol formats
#[derive(Debug)]
pub enum ProtocolFormat {
@@ -17,6 +23,7 @@ pub struct ProtocolConfiguration {
protocol_version: i32,
format: ProtocolFormat,
session_token: Option<String>,
ready_payload_fields: ReadyPayloadFields,
}
impl ProtocolConfiguration {
@@ -25,11 +32,13 @@ impl ProtocolConfiguration {
protocol_version: i32,
format: ProtocolFormat,
session_token: Option<String>,
ready_payload_fields: ReadyPayloadFields,
) -> Self {
Self {
protocol_version,
format,
session_token,
ready_payload_fields,
}
}
@@ -38,22 +47,16 @@ impl ProtocolConfiguration {
match self.format {
ProtocolFormat::Json => {
if let Message::Text(text) = msg {
// Log the error in-case we make a breaking change to the payload
serde_json::from_str(text)
.capture_error()
.map_err(|_| create_error!(UnprocessableEntity))
serde_json::from_str(text).map_err(|_| create_error!(InternalError))
} else {
Err(create_error!(UnprocessableEntity))
Err(create_error!(InternalError))
}
}
ProtocolFormat::Msgpack => {
if let Message::Binary(buf) = msg {
rmp_serde::from_slice(buf)
.capture_error()
.map_err(|_| create_error!(UnprocessableEntity))
rmp_serde::from_slice(buf).map_err(|_| create_error!(InternalError))
} else {
Err(create_error!(UnprocessableEntity))
Err(create_error!(InternalError))
}
}
}
@@ -92,14 +95,8 @@ impl ProtocolConfiguration {
}
/// Get ready payload fields
pub fn get_ready_payload_fields(&self) -> Vec<ReadyPayloadFields> {
vec![
ReadyPayloadFields::Users,
ReadyPayloadFields::Servers,
ReadyPayloadFields::Channels,
ReadyPayloadFields::Members,
ReadyPayloadFields::Emoji,
]
pub fn get_ready_payload_fields(&self) -> &ReadyPayloadFields {
&self.ready_payload_fields
}
}
@@ -130,6 +127,23 @@ impl handshake::server::Callback for WebsocketHandshakeCallback {
let mut protocol_version = 1;
let mut format = ProtocolFormat::Json;
let mut session_token = None;
let mut ready_payload_fields = if params.iter().any(|(k, _)| *k == "ready") {
// If they pass the ready field, set all fields to false
ReadyPayloadFields {
users: false,
servers: false,
channels: false,
members: false,
emojis: false,
voice_states: false,
user_settings: Vec::new(),
channel_unreads: false,
policy_changes: false,
}
} else {
ReadyPayloadFields::default()
};
// Parse and map parameters from key-value to known variables.
for (key, value) in params {
@@ -145,6 +159,31 @@ impl handshake::server::Callback for WebsocketHandshakeCallback {
_ => {}
},
"token" => session_token = Some(value.into()),
"ready" => {
// Re-enable all the fields the client specifies
if let Some(captures) = READY_PAYLOAD_FIELD_REGEX.captures(value) {
if let Some(field) = captures.get(0) {
match field.as_str() {
"users" => ready_payload_fields.users = true,
"servers" => ready_payload_fields.servers = true,
"channels" => ready_payload_fields.channels = true,
"members" => ready_payload_fields.members = true,
"emojis" => ready_payload_fields.emojis = true,
"voice_states" => ready_payload_fields.voice_states = true,
"channel_unreads" => ready_payload_fields.channel_unreads = true,
"user_settings" => {
if let Some(subkey) = captures.get(1) {
ready_payload_fields
.user_settings
.push(subkey.as_str().to_string());
}
}
"policy_changes" => ready_payload_fields.policy_changes = true,
_ => {}
}
}
}
}
_ => {}
}
}
@@ -157,6 +196,7 @@ impl handshake::server::Callback for WebsocketHandshakeCallback {
protocol_version,
format,
session_token,
ready_payload_fields,
})
.is_ok()
{

View File

@@ -4,6 +4,7 @@ use futures::future::join_all;
use revolt_database::{
events::client::{EventV1, ReadyPayloadFields},
util::permissions::DatabasePermissionQuery,
voice::get_channel_voice_state,
Channel, Database, Member, MemberCompositeKey, Presence, RelationshipStatus,
};
use revolt_models::v0;
@@ -17,8 +18,9 @@ use super::state::{Cache, State};
impl Cache {
/// Check whether the current user can view a channel
pub async fn can_view_channel(&self, db: &Database, channel: &Channel) -> bool {
#[allow(deprecated)]
match &channel {
Channel::TextChannel { server, .. } | Channel::VoiceChannel { server, .. } => {
Channel::TextChannel { server, .. } => {
let member = self.members.get(server);
let server = self.servers.get(server);
let mut query =
@@ -95,21 +97,23 @@ impl State {
pub async fn generate_ready_payload(
&mut self,
db: &Database,
fields: Vec<ReadyPayloadFields>,
fields: &ReadyPayloadFields,
) -> Result<EventV1> {
let user = self.clone_user();
self.cache.is_bot = user.bot.is_some();
// Fetch pending policy changes.
let policy_changes = if user.bot.is_some() {
vec![]
let policy_changes = if user.bot.is_some() || !fields.policy_changes {
None
} else {
db.fetch_policy_changes()
.await?
.into_iter()
.filter(|policy| policy.created_time > user.last_acknowledged_policy_change)
.map(Into::into)
.collect()
Some(
db.fetch_policy_changes()
.await?
.into_iter()
.filter(|policy| policy.created_time > user.last_acknowledged_policy_change)
.map(Into::into)
.collect(),
)
};
// Find all relationships to the user.
@@ -168,7 +172,7 @@ impl State {
.await?;
// Fetch customisations.
let emojis = if fields.contains(&ReadyPayloadFields::Emoji) {
let emojis = if fields.emojis {
Some(
db.fetch_emoji_by_parent_ids(
&servers
@@ -176,25 +180,56 @@ impl State {
.map(|x| x.id.to_string())
.collect::<Vec<String>>(),
)
.await?,
.await?
.into_iter()
.map(|emoji| emoji.into())
.collect(),
)
} else {
None
};
// Fetch user settings
let user_settings = if let Some(ReadyPayloadFields::UserSettings(keys)) = fields
.iter()
.find(|e| matches!(e, ReadyPayloadFields::UserSettings(_)))
{
Some(db.fetch_user_settings(&user.id, keys).await?)
let user_settings = if !fields.user_settings.is_empty() {
Some(
db.fetch_user_settings(&user.id, &fields.user_settings)
.await?,
)
} else {
None
};
// Fetch channel unreads
let channel_unreads = if fields.contains(&ReadyPayloadFields::ChannelUnreads) {
Some(db.fetch_unreads(&user.id).await?)
let channel_unreads = if fields.channel_unreads {
Some(
db.fetch_unreads(&user.id)
.await?
.into_iter()
.map(|unread| unread.into())
.collect(),
)
} else {
None
};
let voice_states = if fields.voice_states {
// fetch voice states for all the channels we can see
let mut voice_states = Vec::new();
for channel in channels.iter().filter(|c| {
matches!(
c,
Channel::DirectMessage { .. }
| Channel::Group { .. }
| Channel::TextChannel { voice: Some(_), .. }
)
}) {
if let Ok(Some(voice_state)) = get_channel_voice_state(channel).await {
voice_states.push(voice_state)
}
}
Some(voice_states)
} else {
None
};
@@ -241,30 +276,27 @@ impl State {
}
Ok(EventV1::Ready {
users: if fields.contains(&ReadyPayloadFields::Users) {
Some(users)
} else {
None
},
servers: if fields.contains(&ReadyPayloadFields::Servers) {
users: if fields.users { Some(users) } else { None },
servers: if fields.servers {
Some(servers.into_iter().map(Into::into).collect())
} else {
None
},
channels: if fields.contains(&ReadyPayloadFields::Channels) {
channels: if fields.channels {
Some(channels.into_iter().map(Into::into).collect())
} else {
None
},
members: if fields.contains(&ReadyPayloadFields::Members) {
members: if fields.members {
Some(members.into_iter().map(Into::into).collect())
} else {
None
},
emojis: emojis.map(|vec| vec.into_iter().map(Into::into).collect()),
voice_states,
emojis,
user_settings,
channel_unreads: channel_unreads.map(|vec| vec.into_iter().map(Into::into).collect()),
channel_unreads,
policy_changes,
})
@@ -279,19 +311,14 @@ impl State {
let id = &id.to_string();
for (channel_id, channel) in &self.cache.channels {
match channel {
Channel::TextChannel { server, .. } | Channel::VoiceChannel { server, .. } => {
if server == id {
channel_ids.insert(channel_id.clone());
if channel.server() == Some(id) {
channel_ids.insert(channel_id.clone());
if self.cache.can_view_channel(db, channel).await {
added_channels.push(channel_id.clone());
} else {
removed_channels.push(channel_id.clone());
}
}
if self.cache.can_view_channel(db, channel).await {
added_channels.push(channel_id.clone());
} else {
removed_channels.push(channel_id.clone());
}
_ => {}
}
}
@@ -459,6 +486,7 @@ impl State {
server,
channels,
emojis: _,
voice_states: _,
} => {
self.insert_subscription(id.clone()).await;

View File

@@ -1,7 +1,9 @@
use std::env;
use std::{env, sync::Arc};
use async_std::net::TcpListener;
use revolt_presence::clear_region;
use once_cell::sync::OnceCell;
use revolt_database::voice::VoiceClient;
#[macro_use]
extern crate log;
@@ -12,6 +14,15 @@ pub mod events;
mod database;
mod websocket;
pub static VOICE_CLIENT: OnceCell<Arc<VoiceClient>> = OnceCell::new();
pub fn get_voice_client() -> Arc<VoiceClient> {
VOICE_CLIENT
.get()
.expect("get_voice_client called before set")
.clone()
}
#[async_std::main]
async fn main() {
// Configure requirements for Bonfire.
@@ -24,6 +35,8 @@ async fn main() {
clear_region(None).await;
}
VOICE_CLIENT.set(Arc::new(VoiceClient::from_revolt_config().await)).unwrap();
// Setup a TCP listener to accept WebSocket connections on.
// By default, we bind to port 14703 on all interfaces.
let bind = env::var("HOST").unwrap_or_else(|_| "0.0.0.0:14703".into());

View File

@@ -14,6 +14,7 @@ use futures::{
FutureExt, SinkExt, StreamExt, TryStreamExt,
};
use redis_kiss::{PayloadType, REDIS_PAYLOAD_TYPE, REDIS_URI};
use revolt_config::report_internal_error;
use revolt_database::{
events::{client::EventV1, server::ClientMessage},
iso8601_timestamp::Timestamp,
@@ -26,7 +27,7 @@ use async_std::{
sync::{Mutex, RwLock},
task::spawn,
};
use revolt_result::{create_error, ToRevoltError};
use revolt_result::create_error;
use sentry::Level;
use crate::config::{ProtocolConfiguration, WebsocketHandshakeCallback};
@@ -109,21 +110,21 @@ pub async fn client(db: &'static Database, stream: TcpStream, addr: SocketAddr)
let user_id = state.cache.user_id.clone();
// Notify socket we have authenticated.
if write.send(config.encode(&EventV1::Authenticated)).await.to_internal_error().is_err() {
if report_internal_error!(write.send(config.encode(&EventV1::Authenticated)).await).is_err() {
return;
}
// Download required data to local cache and send Ready payload.
let ready_payload = match state
let ready_payload = match report_internal_error!(
state
.generate_ready_payload(db, config.get_ready_payload_fields())
.await
.to_internal_error()
{
) {
Ok(ready_payload) => ready_payload,
Err(_) => return,
};
if write.send(config.encode(&ready_payload)).await.to_internal_error().is_err() {
if report_internal_error!(write.send(config.encode(&ready_payload)).await).is_err() {
return;
}
@@ -218,15 +219,14 @@ async fn listener(
write: &Mutex<WsWriter>,
) {
let redis_config = RedisConfig::from_url(&REDIS_URI).unwrap();
let subscriber = match fred::types::Builder::from_config(redis_config)
.build_subscriber_client()
.to_internal_error()
{
let subscriber = match report_internal_error!(
fred::types::Builder::from_config(redis_config).build_subscriber_client()
) {
Ok(subscriber) => subscriber,
Err(_) => return,
};
if subscriber.init().await.to_internal_error().is_err() {
if report_internal_error!(subscriber.init().await).is_err() {
return;
}
@@ -249,13 +249,13 @@ async fn listener(
// Check for state changes for subscriptions.
match state.apply_state().await {
SubscriptionStateChange::Reset => {
if subscriber.unsubscribe_all().await.to_internal_error().is_err() {
if report_internal_error!(subscriber.unsubscribe_all().await).is_err() {
break 'out;
}
let subscribed = state.subscribed.read().await;
for id in subscribed.iter() {
if subscriber.subscribe(id).await.to_internal_error().is_err() {
if report_internal_error!(subscriber.subscribe(id).await).is_err() {
break 'out;
}
}
@@ -268,7 +268,7 @@ async fn listener(
#[cfg(debug_assertions)]
info!("{addr:?} unsubscribing from {id}");
if subscriber.unsubscribe(id).await.to_internal_error().is_err() {
if report_internal_error!(subscriber.unsubscribe(id).await).is_err() {
break 'out;
}
}
@@ -277,7 +277,7 @@ async fn listener(
#[cfg(debug_assertions)]
info!("{addr:?} subscribing to {id}");
if subscriber.subscribe(id).await.to_internal_error().is_err() {
if report_internal_error!(subscriber.subscribe(id).await).is_err() {
break 'out;
}
}
@@ -302,7 +302,7 @@ async fn listener(
_ = t2 => {},
message = t1 => {
// Handle incoming events.
let message = match message.to_internal_error() {
let message = match report_internal_error!(message) {
Ok(message) => message,
Err(_) => break 'out
};
@@ -311,15 +311,15 @@ async fn listener(
PayloadType::Json => message
.value
.as_str()
.and_then(|s| serde_json::from_str::<EventV1>(s.as_ref()).to_internal_error().ok()),
.and_then(|s| report_internal_error!(serde_json::from_str::<EventV1>(s.as_ref())).ok()),
PayloadType::Msgpack => message
.value
.as_bytes()
.and_then(|b| rmp_serde::from_slice::<EventV1>(b).to_internal_error().ok()),
.and_then(|b| report_internal_error!(rmp_serde::from_slice::<EventV1>(b)).ok()),
PayloadType::Bincode => message
.value
.as_bytes()
.and_then(|b| bincode::deserialize::<EventV1>(b).to_internal_error().ok()),
.and_then(|b| report_internal_error!(bincode::deserialize::<EventV1>(b)).ok()),
};
let Some(mut event) = event else {
@@ -379,7 +379,7 @@ async fn listener(
}
}
subscriber.quit().await.to_internal_error().ok();
report_internal_error!(subscriber.quit().await).ok();
}
#[allow(clippy::too_many_arguments)]

View File

@@ -0,0 +1,22 @@
[package]
name = "revolt-coalesced"
version = "0.8.9"
edition = "2021"
license = "MIT"
authors = ["Paul Makles <me@insrt.uk>", "Zomatree <me@zomatree.live>"]
description = "Revolt Backend: Coalescion service"
[features]
tokio = ["dep:tokio"]
queue = ["dep:indexmap"]
cache = ["dep:lru"]
default = ["tokio"]
[dependencies]
tokio = { version = "1.47.0", features = ["sync"], optional = true }
indexmap = { version = "*", optional = true }
lru = { version = "*", optional = true }
[dev-dependencies]
tokio = { version = "1.47.0", features = ["rt", "rt-multi-thread", "macros", "time"] }

View File

@@ -0,0 +1,9 @@
MIT License
Copyright (c) 2024 Pawel Makles
Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

View File

@@ -0,0 +1,24 @@
#[derive(Clone, PartialEq, Eq, Debug)]
/// Config values for [`CoalescionService`].
pub struct CoalescionServiceConfig {
/// How many tasks are running at once
pub max_concurrent: Option<usize>,
/// Whether to queue tasks once `max_concurrent` is reached
#[cfg(feature = "queue")]
pub queue_requests: bool,
/// Max amount of tasks in the buffer queue
#[cfg(feature = "queue")]
pub max_queue: Option<usize>,
}
impl Default for CoalescionServiceConfig {
fn default() -> Self {
Self {
max_concurrent: Some(100),
#[cfg(feature = "queue")]
queue_requests: true,
#[cfg(feature = "queue")]
max_queue: Some(100)
}
}
}

View File

@@ -0,0 +1,27 @@
use std::fmt;
#[derive(Clone, Copy, PartialEq, Eq, Debug, Hash)]
/// Coalescion service error.
pub enum Error {
/// Failed to receive the actions return from the channel for unknown reason
RecvError,
/// Reached the `max_concurrent` amount of actions running at once and could not queue the action
MaxConcurrent,
/// Reached the `max_queue` amount of actions in the queue
MaxQueue,
/// Failed to downcast the type to the current type being returned, this will be most likely an ID collision
DowncastError,
}
impl fmt::Display for Error {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Error::RecvError => write!(f, "Unable to receive data from the channel"),
Error::MaxConcurrent => write!(f, "Max number of tasks running at once"),
Error::MaxQueue => write!(f, "Max number of tasks in queue"),
Error::DowncastError => write!(f, "Failed to downcast type, possible key collision with different types")
}
}
}
impl std::error::Error for Error {}

View File

@@ -0,0 +1,39 @@
//! # Coalesced
//!
//! Coalescion service to group, caching and queue duplicate actions.
//! useful for deduplicating web requests, database lookups and other similar resource
//! intensive or rate-limited actions.
//!
//! ## Features
//! - `tokio`: Uses tokio for the async backend, this is currently the only backend.
//! - `queue`: Whether to support queueing requests to only allow X amount of actions running at once.
//! - `cache`: Whether to cache the actions results for future actions with the same id, uses an LRU cache internally.
//!
//! [`CoalescionService`] uses both [`Arc`] and [`RwLock`] internally and can be cheaply cloned to
//! use in your codebase.
//!
//! It is common practice to wrap the service and in your own which delegates the executions to ensure all ids are tracked in one location across your codebase.
//!
//! All values are stored using [`Any`] and must be [`'static`] + [`Send`] + [`Sync`], if there is an id mismatch
//! and a type is wrong the library will return an error, values returned from the service are also
//! wrapped in an [`Arc`] as they are shared to each duplicate action.
//!
//! ## Example:
//! ```rs
//! use revolt_coalesced::CoalescionService;
//!
//! let service = CoalescionService::new();
//!
//! let user_id = "my_user_id";
//! let user = service.execute(user_id, || async move {
//! database.fetch_user(user_id).await.unwrap()
//! }).await;
//! ```
mod config;
mod error;
mod service;
pub use config::CoalescionServiceConfig;
pub use error::Error;
pub use service::CoalescionService;

View File

@@ -0,0 +1,208 @@
use std::{any::Any, collections::HashMap, fmt::Debug, future::Future, hash::Hash, sync::Arc};
use tokio::sync::{
watch::{channel as watch_channel, Receiver},
RwLock,
};
#[cfg(feature = "cache")]
use lru::LruCache;
#[cfg(feature = "queue")]
use indexmap::IndexMap;
use crate::{CoalescionServiceConfig, Error};
#[derive(Debug, Clone)]
#[allow(clippy::type_complexity)]
/// # Coalescion service
///
/// See module description for example usage.
pub struct CoalescionService<Id: Hash + Clone + Eq> {
config: Arc<CoalescionServiceConfig>,
watchers: Arc<RwLock<HashMap<Id, Receiver<Option<Result<Arc<dyn Any + Send + Sync>, Error>>>>>>,
#[cfg(feature = "queue")]
queue: Arc<RwLock<IndexMap<Id, Receiver<Option<Result<Arc<dyn Any + Send + Sync>, Error>>>>>>,
#[cfg(feature = "cache")]
cache: Option<Arc<tokio::sync::Mutex<LruCache<Id, Arc<dyn Any + Send + Sync>>>>>,
}
impl<Id: Hash + Clone + Eq> CoalescionService<Id> {
pub fn new() -> Self {
Default::default()
}
pub fn from_config(config: CoalescionServiceConfig) -> Self {
Self {
config: Arc::new(config),
watchers: Arc::new(RwLock::new(HashMap::new())),
#[cfg(feature = "queue")]
queue: Arc::new(RwLock::new(IndexMap::new())),
#[cfg(feature = "cache")]
cache: None,
}
}
#[cfg(feature = "cache")]
pub fn from_cache(
config: CoalescionServiceConfig,
cache: LruCache<Id, Arc<dyn Any + Send + Sync>>,
) -> Self {
Self {
cache: Some(Arc::new(Mutex::new(cache))),
..Self::from_config(config)
}
}
async fn wait_for<Value: Any + Send + Sync>(
&self,
mut receiver: Receiver<Option<Result<Arc<dyn Any + Send + Sync>, Error>>>,
) -> Result<Arc<Value>, Error> {
receiver
.wait_for(|v| v.is_some())
.await
.map_err(|_| Error::RecvError)
.and_then(|r| r.clone().unwrap())
.and_then(|arc| Arc::downcast(arc).map_err(|_| Error::DowncastError))
}
async fn insert_and_execute<
Value: Send + Sync + 'static,
F: FnOnce() -> Fut,
Fut: Future<Output = Value>,
>(
&self,
id: Id,
func: F,
) -> Result<Arc<Value>, Error> {
let (send, recv) = watch_channel(None);
self.watchers.write().await.insert(id.clone(), recv);
let value = Ok(Arc::new(func().await));
send.send_modify(|opt| {
opt.replace(value.clone().map(|v| v as Arc<dyn Any + Send + Sync>));
});
#[cfg(feature = "cache")]
if let Some(cache) = self.cache.as_ref() {
if let Ok(value) = &value {
cache.lock().await.push(id.clone(), value.clone());
}
};
self.watchers.write().await.remove(&id);
value
}
/// Coalesces an function, the actual function may not run if one with the same id is already running,
/// queued to be ran, or cached, the id should be globally unique for this specific action.
pub async fn execute<
Value: Send + Sync + 'static,
F: FnOnce() -> Fut,
Fut: Future<Output = Value>,
>(
&self,
id: Id,
func: F,
) -> Result<Arc<Value>, Error> {
#[cfg(feature = "cache")]
if let Some(cache) = self.cache.as_ref() {
if let Some(value) = cache.lock().await.get(&id) {
return Arc::downcast::<Value>(value.clone()).map_err(|_| Error::DowncastError);
}
};
let (receiver, length) = {
let watchers = self.watchers.read().await;
let length = watchers.len();
(watchers.get(&id).cloned(), length)
};
if let Some(receiver) = receiver {
self.wait_for(receiver).await
} else {
match self.config.max_concurrent {
Some(max_concurrent) if length >= max_concurrent => {
#[cfg(feature = "queue")]
if self.config.queue_requests {
let (receiver, length) = {
let queue = self.queue.read().await;
(queue.get(&id).cloned(), queue.len())
};
if let Some(receiver) = receiver {
return self.wait_for(receiver).await;
} else {
if self
.config
.max_queue
.is_some_and(|max_queue| max_queue >= length)
{
return Err(Error::MaxQueue);
};
let (send, recv) = watch_channel(None);
self.queue.write().await.insert(id.clone(), recv);
loop {
let length = self.watchers.read().await.len();
if length < max_concurrent {
let first_key = {
let queue = self.queue.read().await;
queue.first().map(|v| v.0).cloned()
};
if first_key == Some(id.clone()) {
self.queue.write().await.shift_remove(&id);
let response = self.insert_and_execute(id, func).await;
send.send_modify(|opt| {
opt.replace(
response
.clone()
.map(|v| v as Arc<dyn Any + Send + Sync>),
);
});
return response;
}
}
}
}
} else {
Err(Error::MaxConcurrent)
}
#[cfg(not(feature = "queue"))]
Err(Error::MaxConcurrent)
}
_ => self.insert_and_execute(id, func).await,
}
}
}
/// Fetches the amount of currently running tasks
pub async fn current_task_count(&self) -> usize {
self.watchers.read().await.len()
}
#[cfg(feature = "queue")]
/// Fetches the current length of the queue
pub async fn current_queue_len(&self) -> usize {
self.queue.read().await.len()
}
}
impl<Id: Hash + Clone + Eq> Default for CoalescionService<Id> {
fn default() -> Self {
Self::from_config(CoalescionServiceConfig::default())
}
}

View File

@@ -1,6 +1,6 @@
[package]
name = "revolt-config"
version = "0.8.8"
version = "0.8.9"
edition = "2021"
license = "MIT"
authors = ["Paul Makles <me@insrt.uk>"]
@@ -37,4 +37,4 @@ sentry = { version = "0.31.5", optional = true }
sentry-anyhow = { version = "0.38.1", optional = true }
# Core
revolt-result = { version = "0.8.8", path = "../result", optional = true }
revolt-result = { version = "0.8.9", path = "../result", optional = true }

View File

@@ -22,6 +22,8 @@ january = "http://local.revolt.chat/january"
voso_legacy = ""
voso_legacy_ws = ""
[hosts.livekit]
[rabbit]
host = "rabbit"
port = 5672
@@ -56,6 +58,8 @@ voso_legacy_token = ""
trust_cloudflare = false
# easypwned endpoint
easypwned = ""
# Tenor API Key
tenor_key = ""
[api.security.captcha]
# hCaptcha configuration
@@ -66,8 +70,13 @@ hcaptcha_sitekey = ""
# Maximum concurrent connections (to proxy server)
max_concurrent_connections = 50
[api.users]
[api.livekit]
# How long to ring devices for when calling in dms/groups, in seconds
call_ring_duration = 30
[api.livekit.nodes]
[api.users]
[pushd]
# this changes the names of the queues to not overlap
@@ -85,6 +94,7 @@ message_queue = "notifications.origin.message"
mass_mention_queue = "notifications.origin.mass_mention" # handles messages that contain role or everyone mentions
fr_accepted_queue = "notifications.ingest.fr_accepted" # friend request accepted
fr_received_queue = "notifications.ingest.fr_received" # friend request received
dm_call_queue = "notifications.ingest.dm_call" # friend request received
generic_queue = "notifications.ingest.generic" # generic messages (title + body)
ack_queue = "notifications.process.ack" # updates badges for apple devices
@@ -228,6 +238,18 @@ message_attachments = 5
# Maximum number of servers the user can create/join
servers = 50
# Maximum audio frequency (Hz) in voice calls
voice_quality = 16000
# Whether the user can use video streams in voice calls
video = true
# Mamimum resolution (width, height) of video streams in voice calls
video_resolution = [1080, 720]
# Minimum and maximum aspect ratio of video streams in voice calls
video_aspect_ratio = [0.3, 2.5]
[features.limits.new_user.file_upload_size_limit]
# Maximum file size limits (in bytes)
attachments = 20_000_000
@@ -255,6 +277,18 @@ message_attachments = 5
# Maximum number of servers the user can create/join
servers = 100
# Maximum audio frequency (Hz) in voice calls
voice_quality = 16000
# Whether the user can use video streams in voice calls
video = true
# Mamimum resolution (width, height) of video streams in voice calls
video_resolution = [1080, 720]
# Minimum and maximum aspect ratio of video streams in voice calls
video_aspect_ratio = [0.3, 2.5]
[features.limits.default.file_upload_size_limit]
# Maximum file size limits (in bytes)
attachments = 20_000_000
@@ -273,7 +307,9 @@ process_message_delay_limit = 5
# Configuration for Sentry error reporting
api = ""
events = ""
voice_ingress = ""
files = ""
proxy = ""
pushd = ""
crond = ""
gifbox = ""

View File

@@ -125,8 +125,7 @@ pub struct Hosts {
pub events: String,
pub autumn: String,
pub january: String,
pub voso_legacy: String,
pub voso_legacy_ws: String,
pub livekit: HashMap<String, String>,
}
#[derive(Deserialize, Debug, Clone)]
@@ -190,6 +189,7 @@ pub struct ApiSecurity {
pub captcha: ApiSecurityCaptcha,
pub trust_cloudflare: bool,
pub easypwned: String,
pub tenor_key: String,
}
#[derive(Deserialize, Debug, Clone)]
@@ -197,6 +197,25 @@ pub struct ApiWorkers {
pub max_concurrent_connections: usize,
}
#[derive(Deserialize, Debug, Clone)]
pub struct ApiLiveKit {
pub call_ring_duration: usize,
pub nodes: HashMap<String, LiveKitNode>,
}
#[derive(Deserialize, Debug, Clone)]
pub struct LiveKitNode {
pub url: String,
pub lat: f64,
pub lon: f64,
pub key: String,
pub secret: String,
// whether to hide the node in the nodes list
#[serde(default)]
pub private: bool,
}
#[derive(Deserialize, Debug, Clone)]
pub struct ApiUsers {
pub early_adopter_cutoff: Option<u64>,
@@ -208,6 +227,7 @@ pub struct Api {
pub smtp: ApiSmtp,
pub security: ApiSecurity,
pub workers: ApiWorkers,
pub livekit: ApiLiveKit,
pub users: ApiUsers,
}
@@ -220,6 +240,7 @@ pub struct Pushd {
// Queues
pub message_queue: String,
pub mass_mention_queue: String,
pub dm_call_queue: String,
pub fr_accepted_queue: String,
pub fr_received_queue: String,
pub generic_queue: String,
@@ -250,6 +271,10 @@ impl Pushd {
self.get_routing_key(self.mass_mention_queue.clone())
}
pub fn get_dm_call_routing_key(&self) -> String {
self.get_routing_key(self.dm_call_queue.clone())
}
pub fn get_fr_accepted_routing_key(&self) -> String {
self.get_routing_key(self.fr_accepted_queue.clone())
}
@@ -317,6 +342,10 @@ pub struct FeaturesLimits {
pub message_length: usize,
pub message_attachments: usize,
pub servers: usize,
pub voice_quality: u32,
pub video: bool,
pub video_resolution: [u32; 2],
pub video_aspect_ratio: [f32; 2],
pub file_upload_size_limit: HashMap<String, usize>,
}
@@ -361,10 +390,12 @@ pub struct Features {
pub struct Sentry {
pub api: String,
pub events: String,
pub voice_ingress: String,
pub files: String,
pub proxy: String,
pub pushd: String,
pub crond: String,
pub gifbox: String,
}
#[derive(Deserialize, Debug, Clone)]

View File

@@ -1,6 +1,6 @@
[package]
name = "revolt-database"
version = "0.8.8"
version = "0.8.9"
edition = "2021"
license = "AGPL-3.0-or-later"
authors = ["Paul Makles <me@insrt.uk>"]
@@ -24,19 +24,19 @@ default = ["mongodb", "async-std-runtime", "tasks"]
[dependencies]
# Core
revolt-config = { version = "0.8.8", path = "../config", features = [
revolt-config = { version = "0.8.9", path = "../config", features = [
"report-macros",
] }
revolt-result = { version = "0.8.8", path = "../result" }
revolt-models = { version = "0.8.8", path = "../models", features = [
revolt-result = { version = "0.8.9", path = "../result" }
revolt-models = { version = "0.8.9", path = "../models", features = [
"validator",
] }
revolt-presence = { version = "0.8.8", path = "../presence" }
revolt-permissions = { version = "0.8.8", path = "../permissions", features = [
revolt-presence = { version = "0.8.9", path = "../presence" }
revolt-permissions = { version = "0.8.9", path = "../permissions", features = [
"serde",
"bson",
] }
revolt-parser = { version = "0.8.8", path = "../parser" }
revolt-parser = { version = "0.8.9", path = "../parser" }
# Utility
log = "0.4"
@@ -96,3 +96,8 @@ authifier = { version = "1.0.15" }
# RabbitMQ
amqprs = { version = "1.7.0" }
# Voice
livekit-api = "0.4.4"
livekit-protocol = "0.4.0"
livekit-runtime = { version = "0.3.1", features = ["tokio"] }

View File

@@ -2,7 +2,8 @@ use std::collections::HashSet;
use crate::events::rabbit::*;
use crate::User;
use amqprs::channel::BasicPublishArguments;
use amqprs::channel::{BasicPublishArguments, ExchangeDeclareArguments};
use amqprs::connection::OpenConnectionArguments;
use amqprs::{channel::Channel, connection::Connection, error::Error as AMQPError};
use amqprs::{BasicProperties, FieldTable};
use revolt_models::v0::PushNotification;
@@ -25,6 +26,35 @@ impl AMQP {
}
}
pub async fn new_auto() -> AMQP {
let config = revolt_config::config().await;
let connection = Connection::open(&OpenConnectionArguments::new(
&config.rabbit.host,
config.rabbit.port,
&config.rabbit.username,
&config.rabbit.password,
))
.await
.expect("Failed to connect to RabbitMQ");
let channel = connection
.open_channel(None)
.await
.expect("Failed to open RabbitMQ channel");
channel
.exchange_declare(
ExchangeDeclareArguments::new(&config.pushd.exchange, "direct")
.durable(true)
.finish(),
)
.await
.expect("Failed to declare exchange");
AMQP::new(connection, channel)
}
pub async fn friend_request_accepted(
&self,
accepted_request_user: &User,
@@ -240,4 +270,50 @@ impl AMQP {
)
.await
}
/// # DM Call Update
/// Used to send an update about a DM call, eg. start or end of a call.
/// Recipients can be used to narrow the scope of recipients, otherwise all recipients will be notified.
/// `ended` refers to the ringing period, not necessarily the call itself.
pub async fn dm_call_updated(
&self,
initiator_id: &str,
channel_id: &str,
started_at: Option<&str>,
ended: bool,
recipients: Option<Vec<String>>,
) -> Result<(), AMQPError> {
let config = revolt_config::config().await;
let payload = InternalDmCallPayload {
payload: DmCallPayload {
initiator_id: initiator_id.to_string(),
channel_id: channel_id.to_string(),
started_at: started_at.map(|f| f.to_string()),
ended,
},
recipients,
};
let payload = to_string(&payload).unwrap();
debug!(
"Sending dm call update payload on channel {}: {}",
config.pushd.get_dm_call_routing_key(),
payload
);
self.channel
.basic_publish(
BasicProperties::default()
.with_content_type("application/json")
.with_persistence(true)
.finish(),
payload.into(),
BasicPublishArguments::new(
&config.pushd.exchange,
&config.pushd.get_dm_call_routing_key(),
),
)
.await
}
}

View File

@@ -35,7 +35,7 @@ pub enum DatabaseInfo {
}
/// Database
#[derive(Clone)]
#[derive(Clone, Debug)]
pub enum Database {
/// Mock database
Reference(ReferenceDb),

View File

@@ -11,6 +11,7 @@ use serde::Serialize;
database_derived!(
/// MongoDB implementation
#[derive(Debug)]
pub struct MongoDb(pub ::mongodb::Client, pub String);
);

View File

@@ -10,7 +10,7 @@ use crate::{
database_derived!(
/// Reference implementation
#[derive(Default)]
#[derive(Default, Debug)]
pub struct ReferenceDb {
pub bots: Arc<Mutex<HashMap<String, Bot>>>,
pub channels: Arc<Mutex<HashMap<String, Channel>>>,

View File

@@ -3,10 +3,7 @@ use revolt_result::Error;
use serde::{Deserialize, Serialize};
use revolt_models::v0::{
AppendMessage, Channel, ChannelUnread, Emoji, FieldsChannel, FieldsMember, FieldsMessage,
FieldsRole, FieldsServer, FieldsUser, FieldsWebhook, Member, MemberCompositeKey, Message,
PartialChannel, PartialMember, PartialMessage, PartialRole, PartialServer, PartialUser,
PartialWebhook, PolicyChange, RemovalIntention, Report, Server, User, UserSettings, Webhook,
AppendMessage, Channel, ChannelUnread, ChannelVoiceState, Emoji, FieldsChannel, FieldsMember, FieldsMessage, FieldsRole, FieldsServer, FieldsUser, FieldsWebhook, Member, MemberCompositeKey, Message, PartialChannel, PartialMember, PartialMessage, PartialRole, PartialServer, PartialUser, PartialUserVoiceState, PartialWebhook, PolicyChange, RemovalIntention, Report, Server, User, UserSettings, UserVoiceState, Webhook
};
use crate::Database;
@@ -20,16 +17,33 @@ pub enum Ping {
}
/// Fields provided in Ready payload
#[derive(PartialEq)]
pub enum ReadyPayloadFields {
Users,
Servers,
Channels,
Members,
Emoji,
#[derive(PartialEq, Debug, Clone, Deserialize)]
pub struct ReadyPayloadFields {
pub users: bool,
pub servers: bool,
pub channels: bool,
pub members: bool,
pub emojis: bool,
pub voice_states: bool,
pub user_settings: Vec<String>,
pub channel_unreads: bool,
pub policy_changes: bool,
}
UserSettings(Vec<String>),
ChannelUnreads,
impl Default for ReadyPayloadFields {
fn default() -> Self {
Self {
users: true,
servers: true,
channels: true,
members: true,
emojis: true,
voice_states: true,
user_settings: Vec::new(),
channel_unreads: false,
policy_changes: true,
}
}
}
/// Protocol Events
@@ -57,13 +71,16 @@ pub enum EventV1 {
members: Option<Vec<Member>>,
#[serde(skip_serializing_if = "Option::is_none")]
emojis: Option<Vec<Emoji>>,
#[serde(skip_serializing_if = "Option::is_none")]
voice_states: Option<Vec<ChannelVoiceState>>,
#[serde(skip_serializing_if = "Option::is_none")]
user_settings: Option<UserSettings>,
#[serde(skip_serializing_if = "Option::is_none")]
channel_unreads: Option<Vec<ChannelUnread>>,
policy_changes: Vec<PolicyChange>,
#[serde(skip_serializing_if = "Option::is_none")]
policy_changes: Option<Vec<PolicyChange>>,
},
/// Ping response
@@ -122,6 +139,7 @@ pub enum EventV1 {
server: Server,
channels: Vec<Channel>,
emojis: Vec<Emoji>,
voice_states: Vec<ChannelVoiceState>
},
/// Update existing server
@@ -144,7 +162,13 @@ pub enum EventV1 {
},
/// User joins server
ServerMemberJoin { id: String, user: String },
ServerMemberJoin {
id: String,
// Deprecated: use member.id.user
#[deprecated = "Use member.id.user instead"]
user: String,
member: Member,
},
/// User left server
ServerMemberLeave {
@@ -248,6 +272,31 @@ pub enum EventV1 {
/// Auth events
Auth(AuthifierEvent),
/// Voice events
VoiceChannelJoin {
id: String,
state: UserVoiceState,
},
VoiceChannelLeave {
id: String,
user: String,
},
VoiceChannelMove {
user: String,
from: String,
to: String,
state: UserVoiceState
},
UserVoiceStateUpdate {
id: String,
channel_id: String,
data: PartialUserVoiceState,
},
UserMoveVoiceChannel {
node: String,
token: String
}
}
impl EventV1 {

View File

@@ -37,6 +37,20 @@ pub struct GenericPayload {
pub user: User,
}
#[derive(Serialize, Deserialize, Clone)]
pub struct DmCallPayload {
pub initiator_id: String,
pub channel_id: String,
pub started_at: Option<String>,
pub ended: bool,
}
#[derive(Serialize, Deserialize, Clone)]
pub struct InternalDmCallPayload {
pub payload: DmCallPayload,
pub recipients: Option<Vec<String>>,
}
#[derive(Serialize, Deserialize)]
#[serde(tag = "type", content = "data")]
#[allow(clippy::large_enum_variant)]
@@ -46,6 +60,7 @@ pub enum PayloadKind {
FRReceived(FRReceivedPayload),
BadgeUpdate(usize),
Generic(GenericPayload),
DmCallStartEnd(DmCallPayload),
}
#[derive(Serialize, Deserialize)]

View File

@@ -1,8 +1,8 @@
use serde::Deserialize;
use serde::{Serialize, Deserialize};
use super::client::Ping;
#[derive(Deserialize, Debug)]
#[derive(Serialize, Deserialize, Debug)]
#[serde(tag = "type")]
pub enum ClientMessage {
Authenticate { token: String },

View File

@@ -112,6 +112,9 @@ pub mod tasks;
mod amqp;
pub use amqp::amqp::AMQP;
pub mod voice;
/// Utility function to check if a boolean value is false
pub fn if_false(t: &bool) -> bool {
!t

View File

@@ -9,14 +9,13 @@ use crate::{
bson::{doc, from_bson, from_document, to_document, Bson, DateTime, Document},
options::FindOptions,
},
AbstractChannels, AbstractServers, Channel, Invite, MongoDb, User, DISCRIMINATOR_SEARCH_SPACE,
AbstractServers, Invite, MongoDb, User, DISCRIMINATOR_SEARCH_SPACE,
};
use bson::{oid::ObjectId, to_bson};
use futures::StreamExt;
use iso8601_timestamp::Timestamp;
use rand::seq::SliceRandom;
use revolt_permissions::DEFAULT_WEBHOOK_PERMISSIONS;
use revolt_result::{Error, ErrorType};
use revolt_permissions::{ChannelPermission, DEFAULT_WEBHOOK_PERMISSIONS};
use serde::{Deserialize, Serialize};
use unicode_segmentation::UnicodeSegmentation;
@@ -26,7 +25,7 @@ struct MigrationInfo {
revision: i32,
}
pub const LATEST_REVISION: i32 = 42; // MUST BE +1 to last migration
pub const LATEST_REVISION: i32 = 48; // MUST BE +1 to last migration
pub async fn migrate_database(db: &MongoDb) {
let migrations = db.col::<Document>("migrations");
@@ -914,6 +913,7 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
}
if revision <= 26 {
// Need to migrate fields on attachments, change `user_id`, `object_id`, etc to `parent`.
info!("Running migration [revision 26 / 15-05-2024]: fix invites being incorrectly serialized with wrong enum tagging.");
auto_derived!(
@@ -1080,6 +1080,14 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
channel_id: String,
}
#[allow(clippy::enum_variant_names)]
#[derive(serde::Serialize, serde::Deserialize)]
enum Channel {
Group { owner: String },
TextChannel { server: String },
VoiceChannel { server: String }
}
let webhooks = db
.db()
.collection::<WebhookShell>("channel_webhooks")
@@ -1091,8 +1099,8 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
.await;
for webhook in webhooks {
match db.fetch_channel(&webhook.channel_id).await {
Ok(channel) => {
match db.col::<Channel>("channels").find_one(doc! { "_id": &webhook.channel_id }).await.unwrap() {
Some(channel) => {
let creator_id = match channel {
Channel::Group { owner, .. } => owner,
Channel::TextChannel { server, .. }
@@ -1100,7 +1108,6 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
let server = db.fetch_server(&server).await.expect("server");
server.owner
}
_ => unreachable!("not server or group channel!"),
};
db.db()
@@ -1118,17 +1125,13 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
.await
.expect("update webhook");
}
Err(Error {
error_type: ErrorType::NotFound,
..
}) => {
None => {
db.db()
.collection::<WebhookShell>("channel_webhooks")
.delete_one(doc! { "_id": webhook._id })
.await
.expect("failed to delete invalid webhook");
}
Err(err) => panic!("{err:?}"),
}
}
}
@@ -1169,9 +1172,9 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
.expect("failed to update users");
}
if revision <= 41 {
if revision <= 43 {
info!(
"Running migration [revision 41 / 05-06-2025]: convert role ranks to uniform numbers."
"Running migration [revision 43 / 05-06-2025]: convert role ranks to uniform numbers."
);
#[derive(Serialize, Deserialize, Clone)]
@@ -1226,6 +1229,41 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
}
}
if revision <= 46 {
info!("Running migration [revision 46 / 29-04-2025]: Convert all `VoiceChannel`'s into `TextChannel`");
db.col::<Document>("channels")
.update_many(
doc! { "channel_type": "VoiceChannel" },
doc! {
"$set": {
"channel_type": "TextChannel",
"voice": {}
}
}
)
.await
.expect("Failed to update voice channels");
};
if revision <= 47 {
info!("Running migration [revision 47 / 29-04-2025]: Add Video to default permissions");
db.col::<Document>("servers")
.update_many(
doc! { },
doc! {
"$bit": {
"default_permissions": {
"or": ChannelPermission::Video as i64
},
}
}
)
.await
.expect("Failed to update default_permissions");
};
// Reminder to update LATEST_REVISION when adding new migrations.
LATEST_REVISION.max(revision)
}

View File

@@ -69,7 +69,7 @@ impl Invite {
creator: creator.id.clone(),
channel: id.clone(),
}),
Channel::TextChannel { id, server, .. } | Channel::VoiceChannel { id, server, .. } => {
Channel::TextChannel { id, server, .. } => {
Ok(Invite::Server {
code,
creator: creator.id.clone(),

View File

@@ -1,4 +1,5 @@
use std::collections::HashMap;
#![allow(deprecated)]
use std::{borrow::Cow, collections::HashMap};
use revolt_config::config;
use revolt_models::v0::{self, MessageAuthor};
@@ -8,8 +9,7 @@ use serde::{Deserialize, Serialize};
use ulid::Ulid;
use crate::{
events::client::EventV1, Database, File, PartialServer,
Server, SystemMessage, User, AMQP,
events::client::EventV1, Database, File, PartialServer, Server, SystemMessage, User, AMQP,
};
#[cfg(feature = "mongodb")]
@@ -106,38 +106,18 @@ auto_derived!(
/// Whether this channel is marked as not safe for work
#[serde(skip_serializing_if = "crate::if_false", default)]
nsfw: bool,
/// Voice Information for when this channel is also a voice channel
#[serde(skip_serializing_if = "Option::is_none")]
voice: Option<VoiceInformation>,
},
/// Voice channel belonging to a server
VoiceChannel {
/// Unique Id
#[serde(rename = "_id")]
id: String,
/// Id of the server this channel belongs to
server: String,
}
/// Display name of the channel
name: String,
#[serde(skip_serializing_if = "Option::is_none")]
/// Channel description
description: Option<String>,
/// Custom icon attachment
#[serde(skip_serializing_if = "Option::is_none")]
icon: Option<File>,
/// Default permissions assigned to users in this channel
#[serde(skip_serializing_if = "Option::is_none")]
default_permissions: Option<OverrideField>,
/// Permissions assigned based on role to this channel
#[serde(
default = "HashMap::<String, OverrideField>::new",
skip_serializing_if = "HashMap::<String, OverrideField>::is_empty"
)]
role_permissions: HashMap<String, OverrideField>,
/// Whether this channel is marked as not safe for work
#[serde(skip_serializing_if = "crate::if_false", default)]
nsfw: bool,
},
#[derive(Default)]
pub struct VoiceInformation {
/// Maximium amount of users allowed in the voice channel at once
#[serde(skip_serializing_if = "Option::is_none")]
pub max_users: Option<usize>,
}
);
@@ -164,6 +144,8 @@ auto_derived!(
pub default_permissions: Option<OverrideField>,
#[serde(skip_serializing_if = "Option::is_none")]
pub last_message_id: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub voice: Option<VoiceInformation>,
}
/// Optional fields on channel object
@@ -171,6 +153,7 @@ auto_derived!(
Description,
Icon,
DefaultPermissions,
Voice,
}
);
@@ -222,16 +205,19 @@ impl Channel {
default_permissions: None,
role_permissions: HashMap::new(),
nsfw: data.nsfw.unwrap_or(false),
voice: data.voice.map(|voice| voice.into()),
},
v0::LegacyServerChannelType::Voice => Channel::VoiceChannel {
v0::LegacyServerChannelType::Voice => Channel::TextChannel {
id: id.clone(),
server: server.id.to_owned(),
name: data.name,
description: data.description,
icon: None,
last_message_id: None,
default_permissions: None,
role_permissions: HashMap::new(),
nsfw: data.nsfw.unwrap_or(false),
voice: Some(data.voice.unwrap_or_default().into()),
},
};
@@ -432,8 +418,28 @@ impl Channel {
Channel::DirectMessage { id, .. }
| Channel::Group { id, .. }
| Channel::SavedMessages { id, .. }
| Channel::TextChannel { id, .. }
| Channel::VoiceChannel { id, .. } => id,
| Channel::TextChannel { id, .. } => id,
}
}
/// Clone this channel's server id
pub fn server(&self) -> Option<&str> {
match self {
Channel::TextChannel { server, .. } => Some(server),
_ => None,
}
}
/// Gets this channel's voice information
pub fn voice(&self) -> Option<Cow<VoiceInformation>> {
match self {
Self::DirectMessage { .. } | Self::Group { .. } => {
Some(Cow::Owned(VoiceInformation::default()))
}
Self::TextChannel {
voice: Some(voice), ..
} => Some(Cow::Borrowed(voice)),
_ => None,
}
}
@@ -450,12 +456,6 @@ impl Channel {
server,
role_permissions,
..
}
| Channel::VoiceChannel {
id,
server,
role_permissions,
..
} => {
db.set_channel_role_permission(id, role_id, permissions)
.await?;
@@ -502,7 +502,7 @@ impl Channel {
clear: remove.into_iter().map(|v| v.into()).collect(),
}
.p(match self {
Self::TextChannel { server, .. } | Self::VoiceChannel { server, .. } => server.clone(),
Self::TextChannel { server, .. } => server.clone(),
_ => id,
})
.await;
@@ -514,17 +514,13 @@ impl Channel {
pub fn remove_field(&mut self, field: &FieldsChannel) {
match field {
FieldsChannel::Description => match self {
Self::Group { description, .. }
| Self::TextChannel { description, .. }
| Self::VoiceChannel { description, .. } => {
Self::Group { description, .. } | Self::TextChannel { description, .. } => {
description.take();
}
_ => {}
},
FieldsChannel::Icon => match self {
Self::Group { icon, .. }
| Self::TextChannel { icon, .. }
| Self::VoiceChannel { icon, .. } => {
Self::Group { icon, .. } | Self::TextChannel { icon, .. } => {
icon.take();
}
_ => {}
@@ -533,15 +529,17 @@ impl Channel {
Self::TextChannel {
default_permissions,
..
}
| Self::VoiceChannel {
default_permissions,
..
} => {
default_permissions.take();
}
_ => {}
},
FieldsChannel::Voice => match self {
Self::TextChannel { voice, .. } => {
voice.take();
}
_ => {}
},
}
}
@@ -553,6 +551,7 @@ impl Channel {
}
/// Apply partial channel to channel
#[allow(deprecated)]
pub fn apply_options(&mut self, partial: PartialChannel) {
match self {
Self::SavedMessages { .. } => {}
@@ -601,15 +600,7 @@ impl Channel {
nsfw,
default_permissions,
role_permissions,
..
}
| Self::VoiceChannel {
name,
description,
icon,
nsfw,
default_permissions,
role_permissions,
voice,
..
} => {
if let Some(v) = partial.name {
@@ -635,6 +626,10 @@ impl Channel {
if let Some(v) = partial.default_permissions {
default_permissions.replace(v);
}
if let Some(v) = partial.voice {
voice.replace(v);
}
}
}
}
@@ -777,6 +772,7 @@ impl IntoDocumentPath for FieldsChannel {
FieldsChannel::Description => "description",
FieldsChannel::Icon => "icon",
FieldsChannel::DefaultPermissions => "default_permissions",
FieldsChannel::Voice => "voice",
})
}
}

View File

@@ -184,7 +184,7 @@ impl AbstractChannels for MongoDb {
async fn delete_channel(&self, channel: &Channel) -> Result<()> {
let id = channel.id().to_string();
let server_id = match channel {
Channel::TextChannel { server, .. } | Channel::VoiceChannel { server, .. } => {
Channel::TextChannel { server, .. } => {
Some(server)
}
_ => None,

View File

@@ -94,9 +94,6 @@ impl AbstractChannels for ReferenceDb {
match &mut channel {
Channel::TextChannel {
role_permissions, ..
}
| Channel::VoiceChannel {
role_permissions, ..
} => {
if role_permissions.get(role_id).is_some() {
role_permissions.remove(role_id);

View File

@@ -1,7 +1,7 @@
use bson::to_document;
use bson::Document;
use revolt_config::report_internal_error;
use revolt_result::Result;
use revolt_result::ToRevoltError;
use crate::File;
use crate::FileUsedFor;
@@ -106,7 +106,7 @@ impl AbstractAttachments for MongoDb {
},
doc! {
"$set": {
"used_for": to_document(&used_for).to_internal_error()?,
"used_for": report_internal_error!(to_document(&used_for))?,
"uploader_id": uploader_id
}
},

View File

@@ -10,6 +10,7 @@ use revolt_models::v0::{
use revolt_permissions::{calculate_channel_permissions, ChannelPermission, PermissionValue};
use revolt_result::{ErrorType, Result};
use ulid::Ulid;
use validator::Validate;
use crate::{
events::client::EventV1,
@@ -113,6 +114,11 @@ auto_derived!(
MessagePinned { id: String, by: String },
#[serde(rename = "message_unpinned")]
MessageUnpinned { id: String, by: String },
#[serde(rename = "call_started")]
CallStarted {
by: String,
finished_at: Option<Timestamp>,
},
}
/// Name and / or avatar override information
@@ -325,9 +331,7 @@ impl Message {
}
let server_id = match channel {
Channel::TextChannel { ref server, .. } | Channel::VoiceChannel { ref server, .. } => {
Some(server.clone())
}
Channel::TextChannel { ref server, .. } => Some(server.clone()),
_ => None,
};
@@ -477,6 +481,7 @@ impl Message {
// Validate the mentions go to users in the channel/server
if !user_mentions.is_empty() {
#[allow(deprecated)]
match channel {
Channel::DirectMessage { ref recipients, .. }
| Channel::Group { ref recipients, .. } => {
@@ -484,32 +489,34 @@ impl Message {
user_mentions.retain(|m| recipients_hash.contains(m));
role_mentions.clear();
}
Channel::TextChannel { ref server, .. }
| Channel::VoiceChannel { ref server, .. } => {
Channel::TextChannel { ref server, .. }=> {
let mentions_vec = Vec::from_iter(user_mentions.iter().cloned());
let valid_members = db.fetch_members(server.as_str(), &mentions_vec[..]).await?;
let valid_members = db.fetch_members(server.as_str(), &mentions_vec[..]).await;
if let Ok(valid_members) = valid_members {
let valid_mentions = HashSet::<&String, RandomState>::from_iter(
valid_members.iter().map(|m| &m.id.user),
);
let valid_mentions = HashSet::<&String, RandomState>::from_iter(
valid_members.iter().map(|m| &m.id.user),
);
user_mentions.retain(|m| valid_mentions.contains(m)); // quick pass, validate mentions are in the server
user_mentions.retain(|m| valid_mentions.contains(m)); // quick pass, validate mentions are in the server
if !user_mentions.is_empty() {
// if there are still mentions, drill down to a channel-level
let member_channel_view_perms =
BulkDatabasePermissionQuery::from_server_id(db, server)
.await
.channel(&channel)
.members(&valid_members)
.members_can_see_channel()
.await;
if !user_mentions.is_empty() {
// if there are still mentions, drill down to a channel-level
let member_channel_view_perms =
BulkDatabasePermissionQuery::from_server_id(db, server)
.await
.channel(&channel)
.members(&valid_members)
.members_can_see_channel()
.await;
user_mentions
.retain(|m| *member_channel_view_perms.get(m).unwrap_or(&false));
user_mentions
.retain(|m| *member_channel_view_perms.get(m).unwrap_or(&false));
}
} else {
revolt_config::capture_error(&valid_members.unwrap_err());
return Err(create_error!(InternalError));
}
}
Channel::SavedMessages { .. } => {
user_mentions.clear();
@@ -674,7 +681,6 @@ impl Message {
)
.await?;
if !self.has_suppressed_notifications()
&& (self.mentions.is_some() || self.contains_mass_push_mention())
{
@@ -713,6 +719,12 @@ impl Message {
/// Create text embed from sendable embed
pub async fn create_embed(&self, db: &Database, embed: SendableEmbed) -> Result<Embed> {
embed.validate().map_err(|error| {
create_error!(FailedValidation {
error: error.to_string()
})
})?;
let media = if let Some(id) = embed.media {
Some(File::use_attachment(db, &id, &self.id, &self.author).await?)
} else {
@@ -784,7 +796,7 @@ impl Message {
query: MessageQuery,
perspective: &User,
include_users: Option<bool>,
server_id: Option<String>,
server_id: Option<&str>,
) -> Result<BulkMessageResponse> {
let messages: Vec<v0::Message> = db
.fetch_messages(query)
@@ -827,6 +839,7 @@ impl Message {
v0::SystemMessage::MessageUnpinned { by, .. } => {
users.push(by.clone());
}
v0::SystemMessage::CallStarted { by, .. } => users.push(by.clone()),
}
}
users
@@ -841,7 +854,7 @@ impl Message {
users,
members: if let Some(server_id) = server_id {
Some(
db.fetch_members(&server_id, &user_ids)
db.fetch_members(server_id, &user_ids)
.await?
.into_iter()
.map(Into::into)

View File

@@ -1,12 +1,21 @@
use iso8601_timestamp::Timestamp;
use revolt_permissions::{calculate_channel_permissions, ChannelPermission};
use revolt_result::{create_error, Result};
use crate::voice::get_channel_voice_state;
use crate::{
events::client::EventV1, util::permissions::DatabasePermissionQuery, Channel, Database, File,
Server, SystemMessage, User,
events::client::EventV1, util::permissions::DatabasePermissionQuery, Channel,
Database, File, Server, SystemMessage, User,
};
fn default_true() -> bool {
true
}
fn is_true(x: &bool) -> bool {
*x
}
auto_derived_partial!(
/// Server Member
pub struct Member {
@@ -30,6 +39,17 @@ auto_derived_partial!(
/// Timestamp this member is timed out until
#[serde(skip_serializing_if = "Option::is_none")]
pub timeout: Option<Timestamp>,
/// Whether the member is server-wide voice muted
#[serde(skip_serializing_if = "is_true", default = "default_true")]
pub can_publish: bool,
/// Whether the member is server-wide voice deafened
#[serde(skip_serializing_if = "is_true", default = "default_true")]
pub can_receive: bool,
// This value only exists in the database, not the models.
// If it is not-None, the database layer should return None to member fetching queries.
// pub pending_deletion_at: Option<Timestamp>
},
"PartialMember"
);
@@ -50,6 +70,9 @@ auto_derived!(
Avatar,
Roles,
Timeout,
CanReceive,
CanPublish,
JoinedAt,
}
/// Member removal intention
@@ -69,6 +92,8 @@ impl Default for Member {
avatar: None,
roles: vec![],
timeout: None,
can_publish: true,
can_receive: true,
}
}
}
@@ -90,7 +115,7 @@ impl Member {
return Err(create_error!(AlreadyInServer));
}
let member = Member {
let mut member = Member {
id: MemberCompositeKey {
server: server.id.to_string(),
user: user.id.to_string(),
@@ -98,7 +123,9 @@ impl Member {
..Default::default()
};
db.insert_member(&member).await?;
if let Some(updated) = db.insert_or_merge_member(&member).await? {
member = updated;
}
let should_fetch = channels.is_none();
let mut channels = channels.unwrap_or_default();
@@ -121,9 +148,18 @@ impl Member {
let emojis = db.fetch_emoji_by_parent_id(&server.id).await?;
let mut voice_states = Vec::new();
for channel in &channels {
if let Ok(Some(voice_state)) = get_channel_voice_state(channel).await {
voice_states.push(voice_state)
}
}
EventV1::ServerMemberJoin {
id: server.id.clone(),
user: user.id.clone(),
member: member.clone().into(),
}
.p(server.id.clone())
.await;
@@ -137,6 +173,7 @@ impl Member {
.map(|channel| channel.into())
.collect(),
emojis: emojis.into_iter().map(|emoji| emoji.into()).collect(),
voice_states
}
.private(user.id.clone())
.await;
@@ -186,10 +223,13 @@ impl Member {
pub fn remove_field(&mut self, field: &FieldsMember) {
match field {
FieldsMember::JoinedAt => (),
FieldsMember::Avatar => self.avatar = None,
FieldsMember::Nickname => self.nickname = None,
FieldsMember::Roles => self.roles.clear(),
FieldsMember::Timeout => self.timeout = None,
FieldsMember::CanReceive => self.can_receive = true,
FieldsMember::CanPublish => self.can_publish = true,
}
}
@@ -224,7 +264,7 @@ impl Member {
intention: RemovalIntention,
silent: bool,
) -> Result<()> {
db.delete_member(&self.id).await?;
db.soft_delete_member(&self.id).await?;
EventV1::ServerMemberLeave {
id: self.id.server.to_string(),
@@ -260,3 +300,74 @@ impl Member {
Ok(())
}
}
#[cfg(test)]
mod tests {
use iso8601_timestamp::{Duration, Timestamp};
use revolt_models::v0::DataCreateServer;
use crate::{Member, PartialMember, RemovalIntention, Server, User};
#[async_std::test]
async fn muted_member_rejoin() {
database_test!(|db| async move {
match db {
crate::Database::Reference(_) => return,
crate::Database::MongoDb(_) => (),
}
let owner = User::create(&db, "Server Owner".to_string(), None, None)
.await
.unwrap();
let kickable_user = User::create(&db, "Member".to_string(), None, None)
.await
.unwrap();
let server = Server::create(
&db,
DataCreateServer {
name: "Server".to_string(),
description: None,
nsfw: None,
},
&owner,
false,
)
.await
.unwrap()
.0;
Member::create(&db, &server, &owner, None).await.unwrap();
let mut kickable_member = Member::create(&db, &server, &kickable_user, None)
.await
.unwrap()
.0;
kickable_member
.update(
&db,
PartialMember {
timeout: Some(Timestamp::now_utc() + Duration::minutes(5)),
..Default::default()
},
vec![],
)
.await
.unwrap();
assert!(kickable_member.in_timeout());
kickable_member
.remove(&db, &server, RemovalIntention::Kick, false)
.await
.unwrap();
let kickable_member = Member::create(&db, &server, &kickable_user, None)
.await
.unwrap()
.0;
assert!(kickable_member.in_timeout())
});
}
}

View File

@@ -73,13 +73,13 @@ impl ChunkedServerMembersGenerator {
#[async_trait]
pub trait AbstractServerMembers: Sync + Send {
/// Insert a new server member into the database
async fn insert_member(&self, member: &Member) -> Result<()>;
async fn insert_or_merge_member(&self, member: &Member) -> Result<Option<Member>>;
/// Fetch a server member by their id
async fn fetch_member(&self, server_id: &str, user_id: &str) -> Result<Member>;
/// Fetch all members in a server
async fn fetch_all_members<'a>(&self, server_id: &str) -> Result<Vec<Member>>;
async fn fetch_all_members(&self, server_id: &str) -> Result<Vec<Member>>;
/// Fetch all members in a server as an iterator
async fn fetch_all_members_chunked(
@@ -100,10 +100,10 @@ pub trait AbstractServerMembers: Sync + Send {
) -> Result<ChunkedServerMembersGenerator>;
/// Fetch all memberships for a user
async fn fetch_all_memberships<'a>(&self, user_id: &str) -> Result<Vec<Member>>;
async fn fetch_all_memberships(&self, user_id: &str) -> Result<Vec<Member>>;
/// Fetch multiple members by their ids
async fn fetch_members<'a>(&self, server_id: &str, ids: &'a [String]) -> Result<Vec<Member>>;
async fn fetch_members(&self, server_id: &str, ids: &[String]) -> Result<Vec<Member>>;
/// Fetch member count of a server
async fn fetch_member_count(&self, server_id: &str) -> Result<usize>;
@@ -119,6 +119,14 @@ pub trait AbstractServerMembers: Sync + Send {
remove: Vec<FieldsMember>,
) -> Result<()>;
/// Delete a server member by their id
async fn delete_member(&self, id: &MemberCompositeKey) -> Result<()>;
/// Marks a user as no longer a member of a server, while retaining the database value.
/// This is used to keep information such as timeouts in place, but will remove information such as join date and applied roles.
async fn soft_delete_member(&self, id: &MemberCompositeKey) -> Result<()>;
/// Forcibly delete a server member by their id.
/// This will cancel any pending timeouts or other longer term actions, and they will not be reapplied on rejoin.
async fn force_delete_member(&self, id: &MemberCompositeKey) -> Result<()>;
/// Fetch all members who have been marked for deletion.
async fn remove_dangling_members(&self) -> Result<()>;
}

View File

@@ -1,4 +1,6 @@
use bson::Document;
use futures::StreamExt;
use iso8601_timestamp::Timestamp;
use mongodb::options::ReadConcern;
use revolt_result::Result;
@@ -11,9 +13,42 @@ static COL: &str = "server_members";
#[async_trait]
impl AbstractServerMembers for MongoDb {
/// Insert a new server member into the database
async fn insert_member(&self, member: &Member) -> Result<()> {
query!(self, insert_one, COL, &member).map(|_| ())
/// Insert a new server member (or use the existing member if one is found)
async fn insert_or_merge_member(&self, member: &Member) -> Result<Option<Member>> {
let existing: Result<Option<Document>> = query!(
self,
find_one,
COL,
doc! {
"_id.server": &member.id.server,
"_id.user": &member.id.user,
"pending_deletion_at": {"$exists": true}
}
);
// Update the existing record if it exist, otherwise make a new record
if existing.is_ok_and(|x| x.is_some()) {
self.col::<Member>(COL)
.find_one_and_update(
doc! {
"_id.server": &member.id.server,
"_id.user": &member.id.user,
},
doc! {
"$set": {
"joined_at": member.joined_at.duration_since(Timestamp::UNIX_EPOCH).whole_seconds(),
},
"$unset": {
"pending_deletion_at": ""
}
},
)
.return_document(mongodb::options::ReturnDocument::After)
.await
.map_err(|_| create_database_error!("update_one", COL))
} else {
query!(self, insert_one, COL, &member).map(|_| ())?;
Ok(None)
}
}
/// Fetch a server member by their id
@@ -24,18 +59,20 @@ impl AbstractServerMembers for MongoDb {
COL,
doc! {
"_id.server": server_id,
"_id.user": user_id
"_id.user": user_id,
"pending_deletion_at": {"$exists": false}
}
)?
.ok_or_else(|| create_error!(NotFound))
}
/// Fetch all members in a server
async fn fetch_all_members<'a>(&self, server_id: &str) -> Result<Vec<Member>> {
async fn fetch_all_members(&self, server_id: &str) -> Result<Vec<Member>> {
Ok(self
.col::<Member>(COL)
.find(doc! {
"_id.server": server_id
"_id.server": server_id,
"pending_deletion_at": {"$exists": false}
})
.await
.map_err(|_| create_database_error!("find", COL))?
@@ -139,11 +176,12 @@ impl AbstractServerMembers for MongoDb {
}
/// Fetch all memberships for a user
async fn fetch_all_memberships<'a>(&self, user_id: &str) -> Result<Vec<Member>> {
async fn fetch_all_memberships(&self, user_id: &str) -> Result<Vec<Member>> {
Ok(self
.col::<Member>(COL)
.find(doc! {
"_id.user": user_id
"_id.user": user_id,
"pending_deletion_at": {"$exists": false}
})
.await
.map_err(|_| create_database_error!("find", COL))?
@@ -159,11 +197,12 @@ impl AbstractServerMembers for MongoDb {
}
/// Fetch multiple members by their ids
async fn fetch_members<'a>(&self, server_id: &str, ids: &'a [String]) -> Result<Vec<Member>> {
async fn fetch_members(&self, server_id: &str, ids: &[String]) -> Result<Vec<Member>> {
Ok(self
.col::<Member>(COL)
.find(doc! {
"_id.server": server_id,
"pending_deletion_at": {"$exists": false},
"_id.user": {
"$in": ids
}
@@ -185,7 +224,8 @@ impl AbstractServerMembers for MongoDb {
async fn fetch_member_count(&self, server_id: &str) -> Result<usize> {
self.col::<Member>(COL)
.count_documents(doc! {
"_id.server": server_id
"_id.server": server_id,
"pending_deletion_at": {"$exists": false}
})
.await
.map(|c| c as usize)
@@ -196,7 +236,8 @@ impl AbstractServerMembers for MongoDb {
async fn fetch_server_count(&self, user_id: &str) -> Result<usize> {
self.col::<Member>(COL)
.count_documents(doc! {
"_id.user": user_id
"_id.user": user_id,
"pending_deletion_at": {"$exists": false}
})
.await
.map(|c| c as usize)
@@ -225,8 +266,42 @@ impl AbstractServerMembers for MongoDb {
.map(|_| ())
}
/// Marks a member for deletion.
/// This will remove the record if the user has no pending actions (eg. timeout),
/// otherwise will slate the record for deletion by revolt_crond once the actions expire.
async fn soft_delete_member(&self, id: &MemberCompositeKey) -> Result<()> {
let member = self.fetch_member(&id.server, &id.user).await;
if let Ok(member) = member {
if member.in_timeout() {
self.col::<Document>(COL)
.update_many(
doc! {
"_id.server": &id.server,
"_id.user": &id.user,
},
doc! {
"$set": {"pending_deletion_at": format!("{}", member.timeout.unwrap().format())},
"$unset": {
"joined_at": "",
"avatar": "",
"nickname": "",
"roles": ""
}
},
)
.await
.map(|_| ())
.map_err(|_| create_database_error!("update_many", COL))
} else {
self.force_delete_member(id).await
}
} else {
Err(create_database_error!("fetch_member", COL))
}
}
/// Delete a server member by their id
async fn delete_member(&self, id: &MemberCompositeKey) -> Result<()> {
async fn force_delete_member(&self, id: &MemberCompositeKey) -> Result<()> {
query!(
self,
delete_one,
@@ -238,15 +313,31 @@ impl AbstractServerMembers for MongoDb {
)
.map(|_| ())
}
async fn remove_dangling_members(&self) -> Result<()> {
let now = Timestamp::now_utc();
let date = bson::to_bson(&now).expect("Failed to serialize timestamp");
self.col::<Document>(COL)
.delete_many(doc! {
"pending_deletion_at": {"$lt": date}
})
.await
.map(|_| ())
.map_err(|_| create_database_error!("count_documents", COL))
}
}
impl IntoDocumentPath for FieldsMember {
fn as_path(&self) -> Option<&'static str> {
Some(match self {
FieldsMember::JoinedAt => "joined_at",
FieldsMember::Avatar => "avatar",
FieldsMember::Nickname => "nickname",
FieldsMember::Roles => "roles",
FieldsMember::Timeout => "timeout",
FieldsMember::CanPublish => "is_publishing",
FieldsMember::CanReceive => "is_receiving",
})
}
}

View File

@@ -8,13 +8,13 @@ use super::{AbstractServerMembers, ChunkedServerMembersGenerator};
#[async_trait]
impl AbstractServerMembers for ReferenceDb {
/// Insert a new server member into the database
async fn insert_member(&self, member: &Member) -> Result<()> {
async fn insert_or_merge_member(&self, member: &Member) -> Result<Option<Member>> {
let mut server_members = self.server_members.lock().await;
if server_members.contains_key(&member.id) {
Err(create_database_error!("insert", "member"))
} else {
server_members.insert(member.id.clone(), member.clone());
Ok(())
Ok(None)
}
}
@@ -31,7 +31,7 @@ impl AbstractServerMembers for ReferenceDb {
}
/// Fetch all members in a server
async fn fetch_all_members<'a>(&self, server_id: &str) -> Result<Vec<Member>> {
async fn fetch_all_members(&self, server_id: &str) -> Result<Vec<Member>> {
let server_members = self.server_members.lock().await;
Ok(server_members
.values()
@@ -105,7 +105,7 @@ impl AbstractServerMembers for ReferenceDb {
}
/// Fetch all memberships for a user
async fn fetch_all_memberships<'a>(&self, user_id: &str) -> Result<Vec<Member>> {
async fn fetch_all_memberships(&self, user_id: &str) -> Result<Vec<Member>> {
let server_members = self.server_members.lock().await;
Ok(server_members
.values()
@@ -115,7 +115,7 @@ impl AbstractServerMembers for ReferenceDb {
}
/// Fetch multiple members by their ids
async fn fetch_members<'a>(&self, server_id: &str, ids: &'a [String]) -> Result<Vec<Member>> {
async fn fetch_members(&self, server_id: &str, ids: &[String]) -> Result<Vec<Member>> {
let server_members = self.server_members.lock().await;
Ok(ids
.iter()
@@ -169,8 +169,26 @@ impl AbstractServerMembers for ReferenceDb {
}
}
/// Soft delete a member
async fn soft_delete_member(&self, id: &MemberCompositeKey) -> Result<()> {
let mut server_members = self.server_members.lock().await;
let member = server_members.get_mut(id);
if let Some(member) = member {
if member.in_timeout() {
panic!("Soft deletion is not implemented.")
} else if server_members.remove(id).is_some() {
Ok(())
} else {
Err(create_error!(NotFound))
}
} else {
Err(create_error!(NotFound))
}
}
/// Delete a server member by their id
async fn delete_member(&self, id: &MemberCompositeKey) -> Result<()> {
async fn force_delete_member(&self, id: &MemberCompositeKey) -> Result<()> {
let mut server_members = self.server_members.lock().await;
if server_members.remove(id).is_some() {
Ok(())
@@ -178,4 +196,8 @@ impl AbstractServerMembers for ReferenceDb {
Err(create_error!(NotFound))
}
}
async fn remove_dangling_members(&self) -> Result<()> {
todo!()
}
}

View File

@@ -1,14 +1,20 @@
use axum::{extract::FromRequestParts, http::request::Parts};
use axum::{extract::{FromRef, FromRequestParts}, http::request::Parts};
use revolt_result::{create_error, Error, Result};
use crate::{Database, User};
#[async_trait::async_trait]
impl FromRequestParts<Database> for User {
impl<S> FromRequestParts<S> for User
where
Database: FromRef<S>,
S: Send + Sync
{
type Rejection = Error;
async fn from_request_parts(parts: &mut Parts, db: &Database) -> Result<User> {
async fn from_request_parts(parts: &mut Parts, state: &S) -> Result<User> {
let db = Database::from_ref(state);
if let Some(Ok(bot_token)) = parts.headers.get("x-bot-token").map(|v| v.to_str()) {
let bot = db.fetch_bot_by_token(bot_token).await?;
db.fetch_user(&bot.id).await

View File

@@ -705,8 +705,6 @@ impl User {
duration_days: Option<usize>,
reason: Option<Vec<String>>,
) -> Result<()> {
// TODO: authifier Error should implement Error
let authifier = db.clone().to_authifier().await;
let mut account = authifier
.database

View File

@@ -38,7 +38,6 @@ impl<'r> FromRequest<'r> for User {
if let Some(user) = user {
Outcome::Success(user.clone())
} else {
request.local_cache(|| Some(create_error!(InvalidSession)));
Outcome::Error((Status::Unauthorized, authifier::Error::InvalidSession))
}
}

View File

@@ -14,7 +14,7 @@ use validator::HasLen;
use revolt_result::Result;
use super::DelayedTask;
use crate::Channel::{TextChannel, VoiceChannel};
use crate::Channel::TextChannel;
/// Enumeration of possible events
#[derive(Debug, Eq, PartialEq)]
@@ -191,17 +191,14 @@ pub async fn handle_ack_event(
.await
.expect("Failed to fetch channel from db");
match channel {
TextChannel { server, .. } | VoiceChannel { server, .. } => {
if let Err(err) =
amqp.mass_mention_message_sent(server, mass_mentions).await
{
revolt_config::capture_error(&err);
}
}
_ => {
panic!("Unknown channel type when sending mass mention event");
if let TextChannel { server, .. } = channel {
if let Err(err) =
amqp.mass_mention_message_sent(server, mass_mentions).await
{
revolt_config::capture_error(&err);
}
} else {
panic!("Unknown channel type when sending mass mention event");
}
}
}

View File

@@ -143,6 +143,7 @@ impl From<crate::FieldsWebhook> for FieldsWebhook {
}
impl From<crate::Channel> for Channel {
#[allow(deprecated)]
fn from(value: crate::Channel) -> Self {
match value {
crate::Channel::SavedMessages { id, user } => Channel::SavedMessages { id, user },
@@ -188,6 +189,7 @@ impl From<crate::Channel> for Channel {
default_permissions,
role_permissions,
nsfw,
voice,
} => Channel::TextChannel {
id,
server,
@@ -198,31 +200,14 @@ impl From<crate::Channel> for Channel {
default_permissions,
role_permissions,
nsfw,
},
crate::Channel::VoiceChannel {
id,
server,
name,
description,
icon,
default_permissions,
role_permissions,
nsfw,
} => Channel::VoiceChannel {
id,
server,
name,
description,
icon: icon.map(|file| file.into()),
default_permissions,
role_permissions,
nsfw,
voice: voice.map(|voice| voice.into()),
},
}
}
}
impl From<Channel> for crate::Channel {
#[allow(deprecated)]
fn from(value: Channel) -> crate::Channel {
match value {
Channel::SavedMessages { id, user } => crate::Channel::SavedMessages { id, user },
@@ -268,6 +253,7 @@ impl From<Channel> for crate::Channel {
default_permissions,
role_permissions,
nsfw,
voice,
} => crate::Channel::TextChannel {
id,
server,
@@ -278,25 +264,7 @@ impl From<Channel> for crate::Channel {
default_permissions,
role_permissions,
nsfw,
},
Channel::VoiceChannel {
id,
server,
name,
description,
icon,
default_permissions,
role_permissions,
nsfw,
} => crate::Channel::VoiceChannel {
id,
server,
name,
description,
icon: icon.map(|file| file.into()),
default_permissions,
role_permissions,
nsfw,
voice: voice.map(|voice| voice.into()),
},
}
}
@@ -315,6 +283,7 @@ impl From<crate::PartialChannel> for PartialChannel {
role_permissions: value.role_permissions,
default_permissions: value.default_permissions,
last_message_id: value.last_message_id,
voice: value.voice.map(|voice| voice.into())
}
}
}
@@ -332,6 +301,7 @@ impl From<PartialChannel> for crate::PartialChannel {
role_permissions: value.role_permissions,
default_permissions: value.default_permissions,
last_message_id: value.last_message_id,
voice: value.voice.map(|voice| voice.into())
}
}
}
@@ -342,6 +312,7 @@ impl From<FieldsChannel> for crate::FieldsChannel {
FieldsChannel::Description => crate::FieldsChannel::Description,
FieldsChannel::Icon => crate::FieldsChannel::Icon,
FieldsChannel::DefaultPermissions => crate::FieldsChannel::DefaultPermissions,
FieldsChannel::Voice => crate::FieldsChannel::Voice,
}
}
}
@@ -352,6 +323,7 @@ impl From<crate::FieldsChannel> for FieldsChannel {
crate::FieldsChannel::Description => FieldsChannel::Description,
crate::FieldsChannel::Icon => FieldsChannel::Icon,
crate::FieldsChannel::DefaultPermissions => FieldsChannel::DefaultPermissions,
crate::FieldsChannel::Voice => FieldsChannel::Voice,
}
}
}
@@ -543,6 +515,7 @@ impl From<crate::SystemMessage> for SystemMessage {
crate::SystemMessage::UserRemove { id, by } => Self::UserRemove { id, by },
crate::SystemMessage::MessagePinned { id, by } => Self::MessagePinned { id, by },
crate::SystemMessage::MessageUnpinned { id, by } => Self::MessageUnpinned { id, by },
crate::SystemMessage::CallStarted { by, finished_at } => Self::CallStarted { by, finished_at }
}
}
}
@@ -639,6 +612,8 @@ impl From<crate::Member> for Member {
avatar: value.avatar.map(|f| f.into()),
roles: value.roles,
timeout: value.timeout,
can_publish: value.can_publish,
can_receive: value.can_receive,
}
}
}
@@ -652,6 +627,8 @@ impl From<Member> for crate::Member {
avatar: value.avatar.map(|f| f.into()),
roles: value.roles,
timeout: value.timeout,
can_publish: value.can_publish,
can_receive: value.can_receive,
}
}
}
@@ -665,6 +642,8 @@ impl From<crate::PartialMember> for PartialMember {
avatar: value.avatar.map(|f| f.into()),
roles: value.roles,
timeout: value.timeout,
can_publish: value.can_publish,
can_receive: value.can_receive,
}
}
}
@@ -678,6 +657,8 @@ impl From<PartialMember> for crate::PartialMember {
avatar: value.avatar.map(|f| f.into()),
roles: value.roles,
timeout: value.timeout,
can_publish: value.can_publish,
can_receive: value.can_receive,
}
}
}
@@ -707,6 +688,9 @@ impl From<crate::FieldsMember> for FieldsMember {
crate::FieldsMember::Nickname => FieldsMember::Nickname,
crate::FieldsMember::Roles => FieldsMember::Roles,
crate::FieldsMember::Timeout => FieldsMember::Timeout,
crate::FieldsMember::CanReceive => FieldsMember::CanReceive,
crate::FieldsMember::CanPublish => FieldsMember::CanPublish,
crate::FieldsMember::JoinedAt => FieldsMember::JoinedAt,
}
}
}
@@ -718,6 +702,9 @@ impl From<FieldsMember> for crate::FieldsMember {
FieldsMember::Nickname => crate::FieldsMember::Nickname,
FieldsMember::Roles => crate::FieldsMember::Roles,
FieldsMember::Timeout => crate::FieldsMember::Timeout,
FieldsMember::CanReceive => crate::FieldsMember::CanReceive,
FieldsMember::CanPublish => crate::FieldsMember::CanPublish,
FieldsMember::JoinedAt => crate::FieldsMember::JoinedAt,
}
}
}
@@ -1385,3 +1372,19 @@ impl From<FieldsMessage> for crate::FieldsMessage {
}
}
}
impl From<VoiceInformation> for crate::VoiceInformation {
fn from(value: VoiceInformation) -> Self {
crate::VoiceInformation {
max_users: value.max_users
}
}
}
impl From<crate::VoiceInformation> for VoiceInformation {
fn from(value: crate::VoiceInformation) -> Self {
VoiceInformation {
max_users: value.max_users
}
}
}

View File

@@ -144,10 +144,6 @@ impl<'z> BulkDatabasePermissionQuery<'z> {
Channel::TextChannel {
default_permissions,
..
}
| Channel::VoiceChannel {
default_permissions,
..
} => default_permissions.unwrap_or_default().into(),
_ => Default::default(),
}
@@ -156,16 +152,14 @@ impl<'z> BulkDatabasePermissionQuery<'z> {
}
}
#[allow(dead_code)]
#[allow(dead_code, deprecated)]
fn get_channel_type(&mut self) -> ChannelType {
if let Some(channel) = &self.channel {
match channel {
Channel::DirectMessage { .. } => ChannelType::DirectMessage,
Channel::Group { .. } => ChannelType::Group,
Channel::SavedMessages { .. } => ChannelType::SavedMessages,
Channel::TextChannel { .. } | Channel::VoiceChannel { .. } => {
ChannelType::ServerChannel
}
Channel::TextChannel { .. } => ChannelType::ServerChannel,
}
} else {
ChannelType::Unknown
@@ -179,9 +173,6 @@ impl<'z> BulkDatabasePermissionQuery<'z> {
match channel {
Channel::TextChannel {
role_permissions, ..
}
| Channel::VoiceChannel {
role_permissions, ..
} => role_permissions,
_ => panic!("Not supported for non-server channels"),
}
@@ -208,12 +199,6 @@ async fn calculate_members_permissions<'a>(
role_permissions,
default_permissions,
..
}
| Channel::VoiceChannel {
id,
role_permissions,
default_permissions,
..
} => (id, role_permissions, default_permissions),
_ => panic!("Calculation of member permissions must be done on a server channel"),
};

View File

@@ -0,0 +1,24 @@
use crate::Database;
use revolt_result::Result;
/// Formats a user's name depending on their optional features and location.
/// Factors in server display names and user display names before falling back to username#discriminator.
/// Passing a server in which the user is not a member will result in an Err.
pub async fn format_display_name(
db: &Database,
user_id: &str,
server_id: Option<&str>,
) -> Result<String> {
if let Some(server_id) = server_id {
let member = db.fetch_member(server_id, user_id).await?;
if let Some(nick) = member.nickname {
return Ok(nick);
}
}
let user = db.fetch_user(user_id).await?;
if let Some(display) = user.display_name {
return Ok(display);
}
Ok(format!("{}#{}", user.username, user.discriminator))
}

View File

@@ -113,8 +113,6 @@ impl<'r> FromRequest<'r> for IdempotencyKey {
let idempotency = IdempotencyKey { key };
let mut cache = TOKEN_CACHE.lock().await;
if cache.get(&idempotency.key).is_some() {
request.local_cache(|| Some(create_error!(DuplicateNonce)));
return Outcome::Error((Status::Conflict, create_error!(DuplicateNonce)));
}

View File

@@ -1,6 +1,9 @@
pub mod bridge;
pub mod bulk_permissions;
mod funcs;
pub mod idempotency;
pub mod permissions;
pub mod reference;
pub mod test_fixtures;
pub use funcs::*;

View File

@@ -185,9 +185,26 @@ impl PermissionQuery for DatabasePermissionQuery<'_> {
}
}
async fn do_we_have_publish_overwrites(&mut self) -> bool {
if let Some(member) = &self.member {
member.can_publish
} else {
true
}
}
async fn do_we_have_receive_overwrites(&mut self) -> bool {
if let Some(member) = &self.member {
member.can_receive
} else {
true
}
}
// * For calculating channel permission
/// Get the type of the channel
#[allow(deprecated)]
async fn get_channel_type(&mut self) -> ChannelType {
if let Some(channel) = &self.channel {
match channel {
@@ -199,9 +216,7 @@ impl PermissionQuery for DatabasePermissionQuery<'_> {
Cow::Borrowed(Channel::SavedMessages { .. })
| Cow::Owned(Channel::SavedMessages { .. }) => ChannelType::SavedMessages,
Cow::Borrowed(Channel::TextChannel { .. })
| Cow::Owned(Channel::TextChannel { .. })
| Cow::Borrowed(Channel::VoiceChannel { .. })
| Cow::Owned(Channel::VoiceChannel { .. }) => ChannelType::ServerChannel,
| Cow::Owned(Channel::TextChannel { .. }) => ChannelType::ServerChannel,
}
} else {
ChannelType::Unknown
@@ -225,14 +240,6 @@ impl PermissionQuery for DatabasePermissionQuery<'_> {
| Cow::Owned(Channel::TextChannel {
default_permissions,
..
})
| Cow::Borrowed(Channel::VoiceChannel {
default_permissions,
..
})
| Cow::Owned(Channel::VoiceChannel {
default_permissions,
..
}) => default_permissions.unwrap_or_default().into(),
_ => Default::default(),
}
@@ -250,12 +257,6 @@ impl PermissionQuery for DatabasePermissionQuery<'_> {
})
| Cow::Owned(Channel::TextChannel {
role_permissions, ..
})
| Cow::Borrowed(Channel::VoiceChannel {
role_permissions, ..
})
| Cow::Owned(Channel::VoiceChannel {
role_permissions, ..
}) => {
if let Some(server) = &self.server {
let member_roles = self
@@ -343,11 +344,10 @@ impl PermissionQuery for DatabasePermissionQuery<'_> {
/// (this will only ever be called for server channels, use unimplemented!() for other code paths)
async fn set_server_from_channel(&mut self) {
if let Some(channel) = &self.channel {
#[allow(deprecated)]
match channel {
Cow::Borrowed(Channel::TextChannel { server, .. })
| Cow::Owned(Channel::TextChannel { server, .. })
| Cow::Borrowed(Channel::VoiceChannel { server, .. })
| Cow::Owned(Channel::VoiceChannel { server, .. }) => {
| Cow::Owned(Channel::TextChannel { server, .. }) => {
if let Some(known_server) =
// I'm not sure why I can't just pattern match both at once here?
// It throws some weird error and the provided fix doesn't work :/

View File

@@ -39,7 +39,9 @@ pub async fn load_fixture(db: &Database, input: &str) -> HashMap<String, String>
LoadedFixture::User(user) => db.insert_user(&user).await.unwrap(),
LoadedFixture::Channel(channel) => db.insert_channel(&channel).await.unwrap(),
LoadedFixture::Server(server) => db.insert_server(&server).await.unwrap(),
LoadedFixture::ServerMember(member) => db.insert_member(&member).await.unwrap(),
LoadedFixture::ServerMember(member) => {
db.insert_or_merge_member(&member).await.unwrap();
}
}
}

View File

@@ -0,0 +1,566 @@
use crate::{
events::client::EventV1,
models::{Channel, User},
util::{permissions::DatabasePermissionQuery, reference::Reference},
Database, Server,
};
use iso8601_timestamp::{Duration, Timestamp};
use livekit_protocol::ParticipantPermission;
use redis_kiss::{get_connection as _get_connection, redis::Pipeline, AsyncCommands, Conn};
use revolt_config::FeaturesLimits;
use revolt_models::v0::{self, PartialUserVoiceState, UserVoiceState};
use revolt_permissions::{calculate_channel_permissions, ChannelPermission, PermissionValue};
use revolt_result::{create_error, Result, ToRevoltError};
mod voice_client;
pub use voice_client::VoiceClient;
async fn get_connection() -> Result<Conn> {
_get_connection().await.to_internal_error()
}
pub async fn raise_if_in_voice(user: &User, channel_id: &str) -> Result<()> {
let mut conn = get_connection().await?;
if user.bot.is_some()
// bots can be in as many voice channels as it wants so we just check if its already connected to the one its trying to connect to
&& conn.sismember(format!("vc:{}", &user.id), channel_id)
.await
.to_internal_error()?
{
Err(create_error!(AlreadyConnected))
} else if conn
.scard::<_, u32>(format!("vc:{}", &user.id)) // check if the current vc set is empty
.await
.to_internal_error()?
> 0
{
Err(create_error!(NotConnected))
} else {
Ok(())
}
}
pub async fn set_channel_node(channel: &str, node: &str) -> Result<()> {
get_connection()
.await?
.set(format!("node:{channel}"), node)
.await
.to_internal_error()
}
pub async fn get_channel_node(channel: &str) -> Result<Option<String>> {
get_connection()
.await?
.get(format!("node:{channel}"))
.await
.to_internal_error()
}
pub async fn get_user_voice_channels(user_id: &str) -> Result<Vec<String>> {
get_connection()
.await?
.smembers(format!("vc:{user_id}"))
.await
.to_internal_error()
}
pub async fn set_user_moved_from_voice(
old_channel: &str,
new_channel: &str,
user_id: &str,
) -> Result<()> {
get_connection()
.await?
.set_ex(
format!("moved_from:{user_id}:{old_channel}"),
new_channel,
10,
)
.await
.to_internal_error()
}
pub async fn get_user_moved_from_voice(channel_id: &str, user_id: &str) -> Result<Option<String>> {
get_connection()
.await?
.get_del(format!("moved_from:{user_id}:{channel_id}"))
.await
.to_internal_error()
}
pub async fn set_user_moved_to_voice(
new_channel: &str,
old_channel: &str,
user_id: &str,
) -> Result<()> {
get_connection()
.await?
.set_ex(format!("moved_to:{user_id}:{new_channel}"), old_channel, 10)
.await
.to_internal_error()
}
pub async fn get_user_moved_to_voice(channel_id: &str, user_id: &str) -> Result<Option<String>> {
get_connection()
.await?
.get_del(format!("moved_to:{user_id}:{channel_id}"))
.await
.to_internal_error()
}
pub async fn is_in_voice_channel(user_id: &str, channel_id: &str) -> Result<bool> {
get_connection()
.await?
.sismember(format!("vc:{user_id}"), channel_id)
.await
.to_internal_error()
}
pub async fn get_user_voice_channel_in_server(
user_id: &str,
server_id: &str,
) -> Result<Option<String>> {
let mut conn = get_connection().await?;
let unique_key = format!("{user_id}:{server_id}");
conn.get(&unique_key).await.to_internal_error()
}
pub fn get_allowed_sources(
limits: &FeaturesLimits,
permissions: PermissionValue,
) -> Vec<&'static str> {
let mut allowed_sources = Vec::new();
if permissions.has(ChannelPermission::Speak as u64) {
allowed_sources.push("microphone")
};
if permissions.has(ChannelPermission::Video as u64) && limits.video {
allowed_sources.extend(["camera", "screen_share", "screen_share_audio"]);
};
allowed_sources
}
pub async fn create_voice_state(
channel_id: &str,
server_id: Option<&str>,
user_id: &str,
joined_at: Timestamp,
) -> Result<UserVoiceState> {
let unique_key = format!("{}:{}", &user_id, server_id.unwrap_or(channel_id));
let voice_state = UserVoiceState {
joined_at,
id: user_id.to_string(),
is_receiving: true,
is_publishing: false,
screensharing: false,
camera: false,
};
Pipeline::new()
.sadd(format!("vc_members:{channel_id}"), user_id)
.sadd(format!("vc:{user_id}"), channel_id)
.set(&unique_key, channel_id)
.set(
format!("joined_at:{unique_key}"),
joined_at
.duration_since(Timestamp::UNIX_EPOCH)
.whole_milliseconds() as i64,
)
.set(
format!("is_publishing:{unique_key}"),
voice_state.is_publishing,
)
.set(
format!("is_receiving:{unique_key}"),
voice_state.is_receiving,
)
.set(
format!("screensharing:{unique_key}"),
voice_state.screensharing,
)
.set(format!("camera:{unique_key}"), voice_state.camera)
.query_async::<_, ()>(&mut get_connection().await?.into_inner())
.await
.to_internal_error()?;
Ok(voice_state)
}
pub async fn delete_voice_state(
channel_id: &str,
server_id: Option<&str>,
user_id: &str,
) -> Result<()> {
let unique_key = format!("{}:{}", &user_id, server_id.unwrap_or(channel_id));
Pipeline::new()
.srem(format!("vc_members:{channel_id}"), user_id)
.srem(format!("vc:{user_id}"), channel_id)
.del(&[
format!("joined_at:{unique_key}"),
format!("is_publishing:{unique_key}"),
format!("is_receiving:{unique_key}"),
format!("screensharing:{unique_key}"),
format!("camera:{unique_key}"),
unique_key.clone(),
])
.query_async(&mut get_connection().await?.into_inner())
.await
.to_internal_error()
}
pub async fn delete_channel_voice_state(
channel_id: &str,
server_id: Option<&str>,
user_ids: &[String],
) -> Result<()> {
let parent_id = server_id.unwrap_or(channel_id);
let mut pipeline = Pipeline::new();
pipeline.del(format!("vc_members:{channel_id}"));
for user_id in user_ids {
let unique_key = format!("{user_id}:{parent_id}");
pipeline.srem(format!("vc:{user_id}"), channel_id).del(&[
format!("joined_at:{unique_key}"),
format!("is_publishing:{unique_key}"),
format!("is_receiving:{unique_key}"),
format!("screensharing:{unique_key}"),
format!("camera:{unique_key}"),
unique_key.clone(),
]);
}
pipeline
.query_async(&mut get_connection().await?.into_inner())
.await
.to_internal_error()
}
pub async fn update_voice_state_tracks(
channel_id: &str,
server_id: Option<&str>,
user_id: &str,
added: bool,
track: i32,
) -> Result<PartialUserVoiceState> {
let partial = match track {
/* TrackSource::Unknown */ 0 => PartialUserVoiceState::default(),
/* TrackSource::Camera */
1 => PartialUserVoiceState {
camera: Some(added),
..Default::default()
},
/* TrackSource::Microphone */
2 => PartialUserVoiceState {
is_publishing: Some(added),
..Default::default()
},
/* TrackSource::ScreenShare | TrackSource::ScreenShareAudio */
3 | 4 => PartialUserVoiceState {
screensharing: Some(added),
..Default::default()
},
_ => unreachable!(),
};
update_voice_state(channel_id, server_id, user_id, &partial).await?;
Ok(partial)
}
pub async fn update_voice_state(
channel_id: &str,
server_id: Option<&str>,
user_id: &str,
partial: &PartialUserVoiceState,
) -> Result<()> {
let unique_key = format!("{}:{}", &user_id, server_id.unwrap_or(channel_id));
let mut pipeline = Pipeline::new();
if let Some(camera) = &partial.camera {
pipeline.set(format!("camera:{unique_key}"), camera);
};
if let Some(is_publishing) = &partial.is_publishing {
pipeline.set(format!("is_publishing:{unique_key}"), is_publishing);
}
if let Some(is_receiving) = &partial.is_receiving {
pipeline.set(format!("is_receiving:{unique_key}"), is_receiving);
}
if let Some(screensharing) = &partial.screensharing {
pipeline.set(format!("screensharing:{unique_key}"), screensharing);
}
pipeline
.query_async(&mut get_connection().await?.into_inner())
.await
.to_internal_error()
}
pub async fn get_voice_channel_members(channel_id: &str) -> Result<Option<Vec<String>>> {
get_connection()
.await?
.smembers::<_, Option<Vec<String>>>(format!("vc_members:{channel_id}"))
.await
.to_internal_error()
.map(|opt| opt.and_then(|v| if v.is_empty() { None } else { Some(v) }))
}
pub async fn get_voice_state(
channel_id: &str,
server_id: Option<&str>,
user_id: &str,
) -> Result<Option<UserVoiceState>> {
let unique_key = format!("{}:{}", user_id, server_id.unwrap_or(channel_id));
let (joined_at, is_publishing, is_receiving, screensharing, camera) = get_connection()
.await?
.mget(&[
format!("joined_at:{unique_key}"),
format!("is_publishing:{unique_key}"),
format!("is_receiving:{unique_key}"),
format!("screensharing:{unique_key}"),
format!("camera:{unique_key}"),
])
.await
.to_internal_error()?;
match (
joined_at,
is_publishing,
is_receiving,
screensharing,
camera,
) {
(
Some(joined_at),
Some(is_publishing),
Some(is_receiving),
Some(screensharing),
Some(camera),
) => Ok(Some(v0::UserVoiceState {
joined_at: Timestamp::UNIX_EPOCH
.checked_add(Duration::milliseconds(joined_at))
.unwrap(),
id: user_id.to_string(),
is_receiving,
is_publishing,
screensharing,
camera,
})),
_ => Ok(None),
}
}
pub async fn get_channel_voice_state(channel: &Channel) -> Result<Option<v0::ChannelVoiceState>> {
let members = get_voice_channel_members(channel.id()).await?;
let server = channel.server();
if let Some(members) = members {
let mut participants = Vec::with_capacity(members.len());
for user_id in members {
if let Some(voice_state) = get_voice_state(channel.id(), server, &user_id).await? {
participants.push(voice_state);
} else {
log::info!("Voice state not found but member in voice channel members, removing.");
delete_voice_state(channel.id(), server, &user_id).await?;
}
}
// In case a user voice state failed to be fetched, the vec's capacity will be larger than the length, shrink it
participants.shrink_to_fit();
Ok(Some(v0::ChannelVoiceState {
id: channel.id().to_string(),
participants,
}))
} else {
Ok(None)
}
}
pub async fn move_user(user: &str, from: &str, to: &str) -> Result<()> {
get_connection()
.await?
.smove(
format!("vc-members-{from}"),
format!("vc-members-{to}"),
user,
)
.await
.to_internal_error()
}
pub async fn sync_voice_permissions(
db: &Database,
voice_client: &VoiceClient,
channel: &Channel,
server: Option<&Server>,
role_id: Option<&str>,
) -> Result<()> {
let Some(node) = get_channel_node(channel.id()).await? else {
return Ok(());
};
for user_id in get_voice_channel_members(channel.id())
.await?
.iter()
.flatten()
{
let user = Reference::from_unchecked(user_id).as_user(db).await?;
sync_user_voice_permissions(db, voice_client, &node, &user, channel, server, role_id)
.await?;
}
Ok(())
}
pub async fn sync_user_voice_permissions(
db: &Database,
voice_client: &VoiceClient,
node: &str,
user: &User,
channel: &Channel,
server: Option<&Server>,
role_id: Option<&str>,
) -> Result<()> {
let channel_id = channel.id();
let server_id = server.as_ref().map(|s| s.id.as_str());
let member = match server_id {
Some(server_id) => Some(
Reference::from_unchecked(&user.id)
.as_member(db, server_id)
.await?,
),
None => None,
};
if role_id.is_none_or(|role_id| {
member
.as_ref()
.is_none_or(|member| member.roles.iter().any(|r| r == role_id))
}) {
let Some(voice_state) = get_voice_state(channel_id, server_id, &user.id).await? else {
return Ok(());
};
let mut query = DatabasePermissionQuery::new(db, user)
.channel(channel)
.user(user);
if let (Some(server), Some(member)) = (server, member.as_ref()) {
query = query.member(member).server(server)
}
let permissions = calculate_channel_permissions(&mut query).await;
let limits = user.limits().await;
let mut update_event = PartialUserVoiceState {
id: Some(user.id.clone()),
..Default::default()
};
let before = update_event.clone();
let can_video =
limits.video && permissions.has_channel_permission(ChannelPermission::Video);
let can_speak = permissions.has_channel_permission(ChannelPermission::Speak);
let can_listen = permissions.has_channel_permission(ChannelPermission::Listen);
update_event.camera = voice_state.camera.then_some(can_video);
update_event.screensharing = voice_state.screensharing.then_some(can_video);
update_event.is_publishing = voice_state.is_publishing.then_some(can_speak);
update_voice_state(channel_id, server_id, &user.id, &update_event).await?;
voice_client
.update_permissions(
node,
user,
channel_id,
ParticipantPermission {
can_subscribe: can_listen,
can_publish: can_speak,
can_publish_data: can_speak,
..Default::default()
},
)
.await?;
if update_event != before {
EventV1::UserVoiceStateUpdate {
id: user.id.clone(),
channel_id: channel_id.to_string(),
data: update_event,
}
.p(channel_id.to_string())
.await;
};
};
Ok(())
}
pub async fn set_channel_call_started_system_message(
channel_id: &str,
message_id: &str,
) -> Result<()> {
get_connection()
.await?
.set(format!("call_started_message:{channel_id}"), message_id)
.await
.to_internal_error()
}
pub async fn take_channel_call_started_system_message(channel_id: &str) -> Result<Option<String>> {
get_connection()
.await?
.get_del(format!("call_started_message:{channel_id}"))
.await
.to_internal_error()
}
pub async fn set_call_notification_recipients(
channel_id: &str,
user_id: &str,
recipients: &[String],
) -> Result<()> {
get_connection()
.await?
.set_ex(
format!("call_notification_recipients:{channel_id}-{user_id}"),
recipients,
10,
)
.await
.to_internal_error()
}
pub async fn get_call_notification_recipients(
channel_id: &str,
user_id: &str,
) -> Result<Option<Vec<String>>> {
get_connection()
.await?
.get_del(format!(
"call_notification_recipients:{channel_id}-{user_id}"
))
.await
.to_internal_error()
}

View File

@@ -0,0 +1,156 @@
use crate::{
models::{Channel, User},
Database,
};
use livekit_api::{
access_token::{AccessToken, VideoGrants},
services::room::{CreateRoomOptions, RoomClient as InnerRoomClient, UpdateParticipantOptions},
};
use livekit_protocol::{ParticipantInfo, ParticipantPermission, Room};
use revolt_config::{config, LiveKitNode};
use revolt_permissions::{ChannelPermission, PermissionValue};
use revolt_result::{create_error, Result, ToRevoltError};
use std::{collections::HashMap, time::Duration};
use super::get_allowed_sources;
#[derive(Debug)]
pub struct RoomClient {
pub client: InnerRoomClient,
pub node: LiveKitNode,
}
#[derive(Debug)]
pub struct VoiceClient {
pub rooms: HashMap<String, RoomClient>,
}
impl VoiceClient {
pub fn new(nodes: HashMap<String, LiveKitNode>) -> Self {
Self {
rooms: nodes
.into_iter()
.map(|(name, node)| {
(
name,
RoomClient {
client: InnerRoomClient::with_api_key(
&node.url,
&node.key,
&node.secret,
),
node,
},
)
})
.collect(),
}
}
pub fn is_enabled(&self) -> bool {
!self.rooms.is_empty()
}
pub async fn from_revolt_config() -> Self {
let config = config().await;
Self::new(config.api.livekit.nodes.clone())
}
pub fn get_node(&self, name: &str) -> Result<&RoomClient> {
self.rooms
.get(name)
.ok_or_else(|| create_error!(UnknownNode))
}
pub async fn create_token(
&self,
node: &str,
db: &Database,
user: &User,
permissions: PermissionValue,
channel: &Channel,
) -> Result<String> {
let room = self.get_node(node)?;
let limits = user.limits().await;
let allowed_sources = get_allowed_sources(&limits, permissions);
AccessToken::with_api_key(&room.node.key, &room.node.secret)
.with_name(&format!("{}#{}", user.username, user.discriminator))
.with_identity(&user.id)
.with_metadata(
&serde_json::to_string(&user.clone().into(db, None).await).to_internal_error()?,
)
.with_ttl(Duration::from_secs(10))
.with_grants(VideoGrants {
room_join: true,
can_publish: true,
can_publish_data: false,
can_publish_sources: allowed_sources
.into_iter()
.map(ToString::to_string)
.collect(),
can_subscribe: permissions.has_channel_permission(ChannelPermission::Listen),
room: channel.id().to_string(),
..Default::default()
})
.to_jwt()
.to_internal_error()
}
pub async fn create_room(&self, node: &str, channel: &Channel) -> Result<Room> {
let room = self.get_node(node)?;
room.client
.create_room(
channel.id(),
CreateRoomOptions {
empty_timeout: 5 * 60, // 5 minutes,
..Default::default()
},
)
.await
.to_internal_error()
}
pub async fn update_permissions(
&self,
node: &str,
user: &User,
channel_id: &str,
new_permissions: ParticipantPermission,
) -> Result<ParticipantInfo> {
let room = self.get_node(node)?;
room.client
.update_participant(
channel_id,
&user.id,
UpdateParticipantOptions {
permission: Some(new_permissions),
..Default::default()
},
)
.await
.to_internal_error()
}
pub async fn remove_user(&self, node: &str, user_id: &str, channel_id: &str) -> Result<()> {
let room = self.get_node(node)?;
room.client
.remove_participant(channel_id, user_id)
.await
.to_internal_error()
}
pub async fn delete_room(&self, node: &str, channel_id: &str) -> Result<()> {
let room = self.get_node(node)?;
room.client
.delete_room(channel_id)
.await
.to_internal_error()
}
}

View File

@@ -1,6 +1,6 @@
[package]
name = "revolt-files"
version = "0.8.8"
version = "0.8.9"
edition = "2021"
license = "AGPL-3.0-or-later"
authors = ["Paul Makles <me@insrt.uk>"]
@@ -14,16 +14,16 @@ imagesize = "0.13.0"
tempfile = "3.12.0"
base64 = "0.22.1"
aes-gcm = { version = "0.10.3", features = ["std"] }
aes-gcm = "0.10.3"
typenum = "1.17.0"
aws-config = "1.5.5"
aws-sdk-s3 = { version = "1.46.0", features = ["behavior-version-latest"] }
revolt-config = { version = "0.8.8", path = "../config", features = [
revolt-config = { version = "0.8.9", path = "../config", features = [
"report-macros",
] }
revolt-result = { version = "0.8.8", path = "../result", features = ["sentry"] }
revolt-result = { version = "0.8.9", path = "../result" }
# image processing
jxl-oxide = "0.8.1"

View File

@@ -6,7 +6,7 @@ use aes_gcm::{
};
use image::{DynamicImage, ImageBuffer};
use revolt_config::{config, report_internal_error, FilesS3};
use revolt_result::{create_error, Result, ToRevoltError};
use revolt_result::{create_error, Result};
use aws_sdk_s3::{
config::{Credentials, Region},
@@ -55,12 +55,13 @@ pub async fn fetch_from_s3(bucket_id: &str, path: &str, nonce: &str) -> Result<V
// Send a request for the file
let mut obj =
client.get_object().bucket(bucket_id).key(path).send().await.to_internal_error()?;
report_internal_error!(client.get_object().bucket(bucket_id).key(path).send().await)?;
// Read the file from remote
let mut buf = vec![];
while let Some(bytes) = obj.body.next().await {
buf.write_all(&bytes.to_internal_error()?).to_internal_error()?;
let data = report_internal_error!(bytes)?;
report_internal_error!(buf.write_all(&data))?;
// is there a more efficient way to do this?
// we just want the Vec<u8>
}
@@ -77,7 +78,10 @@ pub async fn fetch_from_s3(bucket_id: &str, path: &str, nonce: &str) -> Result<V
// Decrypt the file
create_cipher(&config.files.encryption_key)
.decrypt_in_place(nonce, b"", &mut buf)
.to_internal_error()?;
.map_err(|_| create_error!(InternalError))?;
// Remove the authentication tag bytes that were added during encryption
buf.truncate(buf.len() - AUTHENTICATION_TAG_SIZE_BYTES);
Ok(buf)
}
@@ -96,17 +100,18 @@ pub async fn upload_to_s3(bucket_id: &str, path: &str, buf: &[u8]) -> Result<Str
// Encrypt the file in place
create_cipher(&config.files.encryption_key)
.encrypt_in_place(&nonce, b"", &mut buf)
.to_internal_error()?;
.map_err(|_| create_error!(InternalError))?;
// Upload the file to remote
client
.put_object()
.bucket(bucket_id)
.key(path)
.body(buf.into())
.send()
.await
.to_internal_error()?;
report_internal_error!(
client
.put_object()
.bucket(bucket_id)
.key(path)
.body(buf.into())
.send()
.await
)?;
Ok(BASE64_STANDARD.encode(nonce))
}
@@ -116,13 +121,14 @@ pub async fn delete_from_s3(bucket_id: &str, path: &str) -> Result<()> {
let config = config().await;
let client = create_client(config.files.s3);
client
.delete_object()
.bucket(bucket_id)
.key(path)
.send()
.await
.to_internal_error()?;
report_internal_error!(
client
.delete_object()
.bucket(bucket_id)
.key(path)
.send()
.await
)?;
Ok(())
}
@@ -142,7 +148,8 @@ pub fn image_size(f: &NamedTempFile) -> Option<(usize, usize)> {
pub fn image_size_vec(v: &[u8], mime: &str) -> Option<(usize, usize)> {
match mime {
"image/svg+xml" => {
let tree = usvg::Tree::from_data(v, &Default::default()).to_internal_error().ok()?;
let tree =
report_internal_error!(usvg::Tree::from_data(v, &Default::default())).ok()?;
let size = tree.size();
Some((size.width() as usize, size.height() as usize))
@@ -217,9 +224,9 @@ pub fn decode_image<R: Read + BufRead + Seek>(reader: &mut R, mime: &str) -> Res
"image/svg+xml" => {
// usvg doesn't support Read trait so copy to buffer
let mut buf = Vec::new();
reader.read_to_end(&mut buf).to_internal_error()?;
report_internal_error!(reader.read_to_end(&mut buf))?;
let tree = usvg::Tree::from_data(&buf, &Default::default()).to_internal_error()?;
let tree = report_internal_error!(usvg::Tree::from_data(&buf, &Default::default()))?;
let size = tree.size();
let mut pixmap = Pixmap::new(size.width() as u32, size.height() as u32)
.ok_or_else(|| create_error!(ImageProcessingFailed))?;
@@ -237,11 +244,10 @@ pub fn decode_image<R: Read + BufRead + Seek>(reader: &mut R, mime: &str) -> Res
))
}
// Check if we can read using image-rs crate
_ => image::ImageReader::new(reader)
.with_guessed_format()
.to_internal_error()?
.decode()
.to_internal_error()
_ => report_internal_error!(report_internal_error!(
image::ImageReader::new(reader).with_guessed_format()
)?
.decode()),
}
}

View File

@@ -1,6 +1,6 @@
[package]
name = "revolt-models"
version = "0.8.8"
version = "0.8.9"
edition = "2021"
license = "MIT"
authors = ["Paul Makles <me@insrt.uk>"]
@@ -20,8 +20,8 @@ default = ["serde", "partials", "rocket"]
[dependencies]
# Core
revolt-config = { version = "0.8.8", path = "../config" }
revolt-permissions = { version = "0.8.8", path = "../permissions" }
revolt-config = { version = "0.8.9", path = "../config" }
revolt-permissions = { version = "0.8.9", path = "../permissions" }
# Utility
regex = "1.11"

View File

@@ -1,4 +1,5 @@
use super::File;
#![allow(deprecated)]
use super::{File, UserVoiceState};
use revolt_permissions::{Override, OverrideField};
use std::collections::{HashMap, HashSet};
@@ -107,44 +108,21 @@ auto_derived!(
serde(skip_serializing_if = "crate::if_false", default)
)]
nsfw: bool,
/// Voice Information for when this channel is also a voice channel
#[cfg_attr(feature = "serde", serde(skip_serializing_if = "Option::is_none"))]
voice: Option<VoiceInformation>,
},
/// Voice channel belonging to a server
VoiceChannel {
/// Unique Id
#[cfg_attr(feature = "serde", serde(rename = "_id"))]
id: String,
/// Id of the server this channel belongs to
server: String,
}
/// Display name of the channel
name: String,
#[cfg_attr(feature = "serde", serde(skip_serializing_if = "Option::is_none"))]
/// Channel description
description: Option<String>,
/// Custom icon attachment
#[cfg_attr(feature = "serde", serde(skip_serializing_if = "Option::is_none"))]
icon: Option<File>,
/// Default permissions assigned to users in this channel
#[cfg_attr(feature = "serde", serde(skip_serializing_if = "Option::is_none"))]
default_permissions: Option<OverrideField>,
/// Permissions assigned based on role to this channel
#[cfg_attr(
feature = "serde",
serde(
default = "HashMap::<String, OverrideField>::new",
skip_serializing_if = "HashMap::<String, OverrideField>::is_empty"
)
)]
role_permissions: HashMap<String, OverrideField>,
/// Whether this channel is marked as not safe for work
#[cfg_attr(
feature = "serde",
serde(skip_serializing_if = "crate::if_false", default)
)]
nsfw: bool,
},
/// Voice information for a channel
#[derive(Default)]
#[cfg_attr(feature = "validator", derive(validator::Validate))]
pub struct VoiceInformation {
/// Maximium amount of users allowed in the voice channel at once
#[cfg_attr(feature = "validator", validate(range(min = 1)))]
#[cfg_attr(feature = "serde", serde(skip_serializing_if = "Option::is_none"))]
pub max_users: Option<usize>,
}
/// Partial representation of a channel
@@ -170,6 +148,8 @@ auto_derived!(
pub default_permissions: Option<OverrideField>,
#[cfg_attr(feature = "serde", serde(skip_serializing_if = "Option::is_none"))]
pub last_message_id: Option<String>,
#[cfg_attr(feature = "serde", serde(skip_serializing_if = "Option::is_none"))]
pub voice: Option<VoiceInformation>,
}
/// Optional fields on channel object
@@ -177,6 +157,7 @@ auto_derived!(
Description,
Icon,
DefaultPermissions,
Voice,
}
/// New webhook information
@@ -205,6 +186,9 @@ auto_derived!(
/// Whether this channel is archived
pub archived: Option<bool>,
/// Voice Information for voice channels
pub voice: Option<VoiceInformation>,
/// Fields to remove from channel
#[cfg_attr(feature = "serde", serde(default))]
pub remove: Vec<FieldsChannel>,
@@ -260,6 +244,10 @@ auto_derived!(
/// Whether this channel is age restricted
#[serde(skip_serializing_if = "Option::is_none")]
pub nsfw: Option<bool>,
/// Voice Information for when this channel is also a voice channel
#[serde(skip_serializing_if = "Option::is_none")]
pub voice: Option<VoiceInformation>,
}
/// New default permissions
@@ -270,7 +258,7 @@ auto_derived!(
permissions: u64,
},
Field {
/// Allow / deny values to set for members in this `TextChannel` or `VoiceChannel`
/// Allow / deny values to set for members in this server channel
permissions: Override,
},
}
@@ -289,9 +277,32 @@ auto_derived!(
}
/// Voice server token response
pub struct LegacyCreateVoiceUserResponse {
pub struct CreateVoiceUserResponse {
/// Token for authenticating with the voice server
token: String,
pub token: String,
/// Url of the livekit server to connect to
pub url: String,
}
/// Voice state for a channel
pub struct ChannelVoiceState {
pub id: String,
/// The states of the users who are connected to the channel
pub participants: Vec<UserVoiceState>,
}
/// Join a voice channel
pub struct DataJoinCall {
/// Name of the node to join
pub node: Option<String>,
/// Whether to force disconnect any other existing voice connections
///
/// Useful for disconnecting on another device and joining on a new.
pub force_disconnect: Option<bool>,
/// Users which should be notified of the call starting
///
/// Only used when the user is the first one connected.
pub recipients: Option<Vec<String>>,
}
);
@@ -302,8 +313,7 @@ impl Channel {
Channel::DirectMessage { id, .. }
| Channel::Group { id, .. }
| Channel::SavedMessages { id, .. }
| Channel::TextChannel { id, .. }
| Channel::VoiceChannel { id, .. } => id,
| Channel::TextChannel { id, .. } => id,
}
}
@@ -315,9 +325,7 @@ impl Channel {
match self {
Channel::DirectMessage { .. } => None,
Channel::SavedMessages { .. } => Some("Saved Messages"),
Channel::TextChannel { name, .. }
| Channel::Group { name, .. }
| Channel::VoiceChannel { name, .. } => Some(name),
Channel::TextChannel { name, .. } | Channel::Group { name, .. } => Some(name),
}
}
}

View File

@@ -132,6 +132,8 @@ auto_derived!(
MessagePinned { id: String, by: String },
#[serde(rename = "message_unpinned")]
MessageUnpinned { id: String, by: String },
#[serde(rename = "call_started")]
CallStarted { by: String, finished_at: Option<Timestamp> },
}
/// Name and / or avatar override information
@@ -445,6 +447,7 @@ impl From<SystemMessage> for String {
}
SystemMessage::MessagePinned { .. } => "Message pinned.".to_string(),
SystemMessage::MessageUnpinned { .. } => "Message unpinned.".to_string(),
SystemMessage::CallStarted { .. } => "Call started.".to_string(),
}
}
}

View File

@@ -7,7 +7,6 @@ mod embeds;
mod emojis;
mod files;
mod messages;
mod onboard;
mod policy_changes;
mod safety_reports;
mod server_bans;
@@ -25,7 +24,6 @@ pub use embeds::*;
pub use emojis::*;
pub use files::*;
pub use messages::*;
pub use onboard::*;
pub use policy_changes::*;
pub use safety_reports::*;
pub use server_bans::*;

View File

@@ -1,15 +0,0 @@
auto_derived!(
/// # New User Data
#[derive(validator::Validate)]
pub struct DataOnboard {
/// New username which will be used to identify the user on the platform
#[validate(length(min = 2, max = 32), regex = "super::RE_USERNAME")]
pub username: String,
}
/// # Onboarding Status
pub struct DataHello {
/// Whether onboarding is required
pub onboarding: bool,
}
);

View File

@@ -1,17 +1,6 @@
use iso8601_timestamp::Timestamp;
auto_derived!(
/// # Report Data
#[derive(validator::Validate)]
pub struct DataReportContent {
/// Content being reported
pub content: ReportedContent,
/// Additional report description
#[validate(length(min = 0, max = 1000))]
#[serde(default)]
pub additional_context: String,
}
/// User-generated platform moderation report
pub struct Report {
/// Unique Id

View File

@@ -31,6 +31,14 @@ pub static RE_COLOUR: Lazy<Regex> = Lazy::new(|| {
Regex::new(r"(?i)^(?:[a-z ]+|var\(--[a-z\d-]+\)|rgba?\([\d, ]+\)|#[a-f0-9]+|(repeating-)?(linear|conic|radial)-gradient\(([a-z ]+|var\(--[a-z\d-]+\)|rgba?\([\d, ]+\)|#[a-f0-9]+|\d+deg)([ ]+(\d{1,3}%|0))?(,[ ]*([a-z ]+|var\(--[a-z\d-]+\)|rgba?\([\d, ]+\)|#[a-f0-9]+)([ ]+(\d{1,3}%|0))?)+\))$").unwrap()
});
fn default_true() -> bool {
true
}
fn is_true(x: &bool) -> bool {
*x
}
auto_derived_partial!(
/// Server Member
pub struct Member {
@@ -57,6 +65,13 @@ auto_derived_partial!(
/// Timestamp this member is timed out until
#[cfg_attr(feature = "serde", serde(skip_serializing_if = "Option::is_none"))]
pub timeout: Option<Timestamp>,
/// Whether the member is server-wide voice muted
#[serde(skip_serializing_if = "is_true", default = "default_true")]
pub can_publish: bool,
/// Whether the member is server-wide voice deafened
#[serde(skip_serializing_if = "is_true", default = "default_true")]
pub can_receive: bool,
},
"PartialMember"
);
@@ -77,6 +92,9 @@ auto_derived!(
Avatar,
Roles,
Timeout,
CanReceive,
CanPublish,
JoinedAt,
}
/// Member removal intention
@@ -123,6 +141,12 @@ auto_derived!(
pub roles: Option<Vec<String>>,
/// Timestamp this member is timed out until
pub timeout: Option<Timestamp>,
/// server-wide voice muted
pub can_publish: Option<bool>,
/// server-wide voice deafened
pub can_receive: Option<bool>,
/// voice channel to move to if already in a voice channel
pub voice_channel: Option<String>,
/// Fields to remove from channel object
#[cfg_attr(feature = "serde", serde(default))]
pub remove: Vec<FieldsMember>,

View File

@@ -1,4 +1,4 @@
use super::{Channel, File, Member, User, RE_COLOUR};
use super::{Channel, File, RE_COLOUR};
use revolt_permissions::{Override, OverrideField};
use std::collections::HashMap;
@@ -293,22 +293,4 @@ auto_derived!(
pub struct DataEditRoleRanks {
pub ranks: Vec<String>,
}
/// # Query Parameters
#[derive(FromForm)]
pub struct OptionsQueryMembers {
/// String to search for
pub query: String,
/// Discourage use of this API
pub experimental_api: bool,
}
/// # Query members by name
pub struct MemberQueryResponse {
/// List of members
pub members: Vec<Member>,
/// List of users
pub users: Vec<User>,
}
);

View File

@@ -1,3 +1,4 @@
use iso8601_timestamp::Timestamp;
use once_cell::sync::Lazy;
use regex::Regex;
@@ -255,12 +256,14 @@ auto_derived!(
pub flags: i32,
}
/// Mutual friends and servers response
/// Mutual friends, servers, groups and DMs response
pub struct MutualResponse {
/// Array of mutual user IDs that both users are friends with
pub users: Vec<String>,
/// Array of mutual server IDs that both users are in
pub servers: Vec<String>,
/// Array of mutual group and dm IDs that both users are in
pub channels: Vec<String>,
}
/// Bot information for if the user is a bot
@@ -275,17 +278,19 @@ auto_derived!(
/// Username and discriminator combo separated by #
pub username: String,
}
);
/// # Username Information
#[derive(Validate)]
pub struct DataChangeUsername {
/// New username
#[validate(length(min = 2, max = 32), regex = "super::RE_USERNAME")]
pub username: String,
/// Current account password
#[validate(length(min = 8, max = 1024))]
pub password: String,
}
auto_derived_partial!(
/// Voice State information for a user
pub struct UserVoiceState {
pub id: String,
pub joined_at: Timestamp,
pub is_receiving: bool,
pub is_publishing: bool,
pub screensharing: bool,
pub camera: bool,
},
"PartialUserVoiceState"
);
pub trait CheckRelationship {

View File

@@ -1,8 +1,9 @@
[package]
name = "revolt-parser"
version = "0.8.8"
version = "0.8.9"
edition = "2021"
license = "AGPL-3.0-or-later"
license = "MIT"
authors = ["Zomatree <me@zomatree.live>", "Paul Makles <me@insrt.uk>"]
description = "Revolt Backend: Message Parser"
[dependencies]

View File

@@ -0,0 +1,9 @@
MIT License
Copyright (c) 2024 Pawel Makles
Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

View File

@@ -4,15 +4,18 @@ use logos::Logos;
#[derive(Debug, Clone, Logos, PartialEq)]
#[logos(skip "\n")]
pub enum MessageToken {
#[logos(subpattern id="[0123456789ABCDEFGHJKMNPQRSTVWXYZ]{26}")]
pub enum MessageToken<'a> {
#[token("\\")]
Escape,
#[regex("(```[^`\n]*)|(``)|`", |lex| lex.slice().to_owned().chars().filter(|&c| c == '`').count())]
#[regex("```[^`\n]*", |_| 3)]
#[regex("``", |_| 2)]
#[regex("`", |_| 1)]
CodeblockMarker(usize),
#[regex("<@[0123456789ABCDEFGHJKMNPQRSTVWXYZ]{26}>", |lex| lex.slice()[2..lex.slice().len() - 1].to_owned())]
UserMention(String),
#[regex("<%[0123456789ABCDEFGHJKMNPQRSTVWXYZ]{26}>", |lex| lex.slice()[2..lex.slice().len() - 1].to_owned())]
RoleMention(String),
#[regex("<@(?&id)>", |lex| &lex.slice()[2..lex.slice().len() - 1])]
UserMention(&'a str),
#[regex("<%(?&id)>", |lex| &lex.slice()[2..lex.slice().len() - 1],)]
RoleMention(&'a str),
#[token("@everyone")]
MentionEveryone,
#[token("@online")]
@@ -27,13 +30,13 @@ pub struct MessageResults {
pub mentions_online: bool
}
struct MessageParserIterator<I> {
struct MessageParserIterator<'a, I> {
inner: I,
temp: VecDeque<MessageToken>
temp: VecDeque<MessageToken<'a>>
}
impl<I: Iterator<Item = MessageToken>> Iterator for MessageParserIterator<I> {
type Item = MessageToken;
impl<'a, I: Iterator<Item = MessageToken<'a>>> Iterator for MessageParserIterator<'a, I> {
type Item = MessageToken<'a>;
fn next(&mut self) -> Option<Self::Item> {
if !self.temp.is_empty() {
@@ -80,8 +83,8 @@ pub fn parse_message(text: &str) -> MessageResults {
match token {
MessageToken::Escape => {}
MessageToken::CodeblockMarker(_) => {},
MessageToken::UserMention(id) => { results.user_mentions.insert(id); },
MessageToken::RoleMention(id) => { results.role_mentions.insert(id); },
MessageToken::UserMention(id) => { results.user_mentions.insert(id.to_string()); },
MessageToken::RoleMention(id) => { results.role_mentions.insert(id.to_string()); },
MessageToken::MentionEveryone => results.mentions_everyone = true,
MessageToken::MentionOnline => results.mentions_online = true,
};
@@ -106,7 +109,7 @@ mod tests {
let output = parse_message_iter("Hello <@01FD58YK5W7QRV5H3D64KTQYX3>.").collect::<Vec<_>>();
assert_eq!(output.len(), 1);
assert_eq!(output[0], MessageToken::UserMention("01FD58YK5W7QRV5H3D64KTQYX3".to_string()));
assert_eq!(output[0], MessageToken::UserMention("01FD58YK5W7QRV5H3D64KTQYX3"));
}
#[test]
@@ -114,7 +117,7 @@ mod tests {
let output = parse_message_iter("Hello <%01FD58YK5W7QRV5H3D64KTQYX3>.").collect::<Vec<_>>();
assert_eq!(output.len(), 1);
assert_eq!(output[0], MessageToken::RoleMention("01FD58YK5W7QRV5H3D64KTQYX3".to_string()));
assert_eq!(output[0], MessageToken::RoleMention("01FD58YK5W7QRV5H3D64KTQYX3"));
}
#[test]
@@ -138,8 +141,8 @@ mod tests {
let output = parse_message_iter("Hello <@01FD58YK5W7QRV5H3D64KTQYX3>, <%01FD58YK5W7QRV5H3D64KTQYX3>, @everyone and @online.").collect::<Vec<_>>();
assert_eq!(output.len(), 4);
assert_eq!(output[0], MessageToken::UserMention("01FD58YK5W7QRV5H3D64KTQYX3".to_string()));
assert_eq!(output[1], MessageToken::RoleMention("01FD58YK5W7QRV5H3D64KTQYX3".to_string()));
assert_eq!(output[0], MessageToken::UserMention("01FD58YK5W7QRV5H3D64KTQYX3"));
assert_eq!(output[1], MessageToken::RoleMention("01FD58YK5W7QRV5H3D64KTQYX3"));
assert_eq!(output[2], MessageToken::MentionEveryone);
assert_eq!(output[3], MessageToken::MentionOnline);
}
@@ -149,8 +152,8 @@ mod tests {
let output = parse_message_iter("<@01FD58YK5W7QRV5H3D64KTQYX3><%01FD58YK5W7QRV5H3D64KTQYX3>@everyone@online").collect::<Vec<_>>();
assert_eq!(output.len(), 4);
assert_eq!(output[0], MessageToken::UserMention("01FD58YK5W7QRV5H3D64KTQYX3".to_string()));
assert_eq!(output[1], MessageToken::RoleMention("01FD58YK5W7QRV5H3D64KTQYX3".to_string()));
assert_eq!(output[0], MessageToken::UserMention("01FD58YK5W7QRV5H3D64KTQYX3"));
assert_eq!(output[1], MessageToken::RoleMention("01FD58YK5W7QRV5H3D64KTQYX3"));
assert_eq!(output[2], MessageToken::MentionEveryone);
assert_eq!(output[3], MessageToken::MentionOnline);
}
@@ -170,15 +173,14 @@ mod tests {
assert_eq!(output.len(), 5);
assert_eq!(output[0], MessageToken::CodeblockMarker(3));
assert_eq!(output[1], MessageToken::UserMention("01FD58YK5W7QRV5H3D64KTQYX3".to_string()));
assert_eq!(output[2], MessageToken::RoleMention("01FD58YK5W7QRV5H3D64KTQYX3".to_string()));
assert_eq!(output[1], MessageToken::UserMention("01FD58YK5W7QRV5H3D64KTQYX3"));
assert_eq!(output[2], MessageToken::RoleMention("01FD58YK5W7QRV5H3D64KTQYX3"));
assert_eq!(output[3], MessageToken::MentionEveryone);
assert_eq!(output[4], MessageToken::MentionOnline);
}
#[test]
fn test_inline_codeblock_no_mentions() {
let output = parse_message_iter("`<@01FD58YK5W7QRV5H3D64KTQYX3><%01FD58YK5W7QRV5H3D64KTQYX3>@everyone@online`").collect::<Vec<_>>();
assert_eq!(output.len(), 2);
@@ -192,12 +194,21 @@ mod tests {
assert_eq!(output.len(), 5);
assert_eq!(output[0], MessageToken::CodeblockMarker(1));
assert_eq!(output[1], MessageToken::UserMention("01FD58YK5W7QRV5H3D64KTQYX3".to_string()));
assert_eq!(output[2], MessageToken::RoleMention("01FD58YK5W7QRV5H3D64KTQYX3".to_string()));
assert_eq!(output[1], MessageToken::UserMention("01FD58YK5W7QRV5H3D64KTQYX3"));
assert_eq!(output[2], MessageToken::RoleMention("01FD58YK5W7QRV5H3D64KTQYX3"));
assert_eq!(output[3], MessageToken::MentionEveryone);
assert_eq!(output[4], MessageToken::MentionOnline);
}
#[test]
fn test_codeblock_with_language_no_mentions() {
let output = parse_message_iter("```rust\n<@01FD58YK5W7QRV5H3D64KTQYX3><%01FD58YK5W7QRV5H3D64KTQYX3>@everyone@online```").collect::<Vec<_>>();
assert_eq!(output.len(), 2);
assert_eq!(output[0], MessageToken::CodeblockMarker(3));
assert_eq!(output[1], MessageToken::CodeblockMarker(3));
}
#[test]
fn test_double_inline_codeblock() {
let output = parse_message_iter("``this should not ping @everyone``").collect::<Vec<_>>();

View File

@@ -1,6 +1,6 @@
[package]
name = "revolt-permissions"
version = "0.8.8"
version = "0.8.9"
edition = "2021"
license = "MIT"
authors = ["Paul Makles <me@insrt.uk>"]
@@ -21,7 +21,7 @@ async-std = { version = "1.8.0", features = ["attributes"] }
[dependencies]
# Core
revolt-result = { version = "0.8.8", path = "../result" }
revolt-result = { version = "0.8.9", path = "../result" }
# Utility
auto_ops = "0.3.0"

View File

@@ -61,6 +61,15 @@ pub async fn calculate_server_permissions<P: PermissionQuery>(query: &mut P) ->
permissions.apply(role_override);
}
if !query.do_we_have_publish_overwrites().await {
permissions.revoke(ChannelPermission::Speak as u64);
permissions.revoke(ChannelPermission::Video as u64);
}
if !query.do_we_have_receive_overwrites().await {
permissions.revoke(ChannelPermission::Listen as u64);
}
if query.are_we_timed_out().await {
permissions.restrict(*ALLOW_IN_TIMEOUT);
}

View File

@@ -89,6 +89,8 @@ pub enum ChannelPermission {
DeafenMembers = 1 << 34,
/// Move members between voice channels
MoveMembers = 1 << 35,
/// Listen to other users
Listen = 1 << 36,
// * Channel permissions two electric boogaloo
/// Mention everyone and online members
@@ -130,7 +132,9 @@ pub static DEFAULT_PERMISSION: Lazy<u64> = Lazy::new(|| {
+ ChannelPermission::SendEmbeds
+ ChannelPermission::UploadFiles
+ ChannelPermission::Connect
+ ChannelPermission::Speak,
+ ChannelPermission::Speak
+ ChannelPermission::Listen
+ ChannelPermission::Video
)
});

View File

@@ -8,10 +8,18 @@ pub use server::*;
pub use user::*;
/// Holds a permission value to manipulate.
#[derive(Clone, Debug)]
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub struct PermissionValue(u64);
impl PermissionValue {
pub fn from_raw(value: u64) -> Self {
Self(value)
}
pub fn into_raw(self) -> u64 {
self.0
}
/// Apply a given override to this value
pub fn apply(&mut self, v: Override) {
self.allow(v.allow);

View File

@@ -39,7 +39,7 @@ pub enum DataPermissionPoly {
permissions: u64,
},
Field {
/// Allow / deny values to set for members in this `TextChannel` or `VoiceChannel`
/// Allow / deny values to set for members in this server channel
permissions: Override,
},
}

View File

@@ -64,6 +64,14 @@ async fn validate_user_permissions() {
unreachable!()
}
async fn do_we_have_publish_overwrites(&mut self) -> bool {
true
}
async fn do_we_have_receive_overwrites(&mut self) -> bool {
true
}
async fn get_channel_type(&mut self) -> ChannelType {
ChannelType::DirectMessage
}
@@ -153,6 +161,14 @@ async fn validate_group_permissions() {
unreachable!()
}
async fn do_we_have_publish_overwrites(&mut self) -> bool {
true
}
async fn do_we_have_receive_overwrites(&mut self) -> bool {
true
}
async fn get_channel_type(&mut self) -> ChannelType {
ChannelType::Group
}
@@ -254,6 +270,14 @@ async fn validate_server_permissions() {
false
}
async fn do_we_have_publish_overwrites(&mut self) -> bool {
true
}
async fn do_we_have_receive_overwrites(&mut self) -> bool {
true
}
async fn get_channel_type(&mut self) -> ChannelType {
ChannelType::ServerChannel
}
@@ -346,6 +370,14 @@ async fn validate_timed_out_member() {
true
}
async fn do_we_have_publish_overwrites(&mut self) -> bool {
true
}
async fn do_we_have_receive_overwrites(&mut self) -> bool {
true
}
async fn get_channel_type(&mut self) -> ChannelType {
ChannelType::ServerChannel
}

View File

@@ -39,6 +39,12 @@ pub trait PermissionQuery {
/// Is our perspective user timed out on this server?
async fn are_we_timed_out(&mut self) -> bool;
/// Is the member muted?
async fn do_we_have_publish_overwrites(&mut self) -> bool;
/// Is the member deafend?
async fn do_we_have_receive_overwrites(&mut self) -> bool;
// * For calculating channel permission
/// Get the type of the channel

View File

@@ -1,6 +1,6 @@
[package]
name = "revolt-presence"
version = "0.8.8"
version = "0.8.9"
edition = "2021"
license = "AGPL-3.0-or-later"
authors = ["Paul Makles <me@insrt.uk>"]
@@ -16,7 +16,7 @@ redis-is-patched = []
async-std = { version = "1.8.0", features = ["attributes"] }
# Config for loading Redis URI
revolt-config = { version = "0.8.8", path = "../config" }
revolt-config = { version = "0.8.9", path = "../config" }
[dependencies]
# Utility

View File

@@ -0,0 +1,26 @@
[package]
name = "revolt-ratelimits"
version = "0.8.9"
edition = "2024"
[features]
rocket = ["dep:rocket", "dep:revolt_rocket_okapi", "revolt-database/rocket-impl"]
axum = ["dep:axum", "revolt-database/axum-impl"]
default = ["rocket", "axum"]
[dependencies]
revolt-database = { version = "0.8.9", path = "../database"}
revolt-result = { version = "0.8.9", path = "../result" }
revolt-config = { version = "0.8.9", path = "../config" }
rocket = { version = "0.5.1", optional = true }
revolt_rocket_okapi = { version = "0.10.0", optional = true }
axum = { version = "0.7.5", optional = true, features = ["macros"] }
serde = { version = "1", features = ["derive"] }
authifier = { version = "1.0.15" }
dashmap = "5.2.0"
async-trait = "0.1.81"
log = "0.4"

View File

@@ -0,0 +1,194 @@
use std::net::SocketAddr;
use async_trait::async_trait;
use axum::{
Json, RequestPartsExt, Router,
body::Body,
extract::{ConnectInfo, FromRef, FromRequestParts, State},
http::{HeaderValue, Request, StatusCode, request::Parts},
middleware::Next,
response::{IntoResponse, Response},
routing::get,
};
use revolt_database::{Database, User};
use revolt_config::config;
use crate::ratelimiter::{RatelimitInformation, Ratelimiter, RequestKind};
#[derive(Clone, Copy)]
pub struct AxumRequestKind;
impl RequestKind for AxumRequestKind {
type R<'a> = Parts;
}
pub type RatelimitStorage = crate::ratelimiter::RatelimitStorage<AxumRequestKind>;
fn to_ip(parts: &Parts) -> String {
parts
.extensions
.get::<ConnectInfo<SocketAddr>>()
.map(|info| info.ip().to_string())
.unwrap_or_default()
}
async fn to_real_ip(parts: &Parts) -> String {
if config().await.api.security.trust_cloudflare {
parts
.headers
.get("CF-Connecting-IP")
.map(|x| x.to_str().unwrap().to_string())
.unwrap_or_else(|| to_ip(parts))
} else {
to_ip(parts)
}
}
#[async_trait]
impl<S: Send + Sync> FromRequestParts<S> for Ratelimiter
where
Database: FromRef<S>,
RatelimitStorage: FromRef<S>,
{
type Rejection = Json<Ratelimiter>;
async fn from_request_parts(parts: &mut Parts, state: &S) -> Result<Self, Self::Rejection> {
if parts
.extensions
.get::<Result<Ratelimiter, Json<Ratelimiter>>>()
.is_none()
{
let storage = RatelimitStorage::from_ref(state);
let identifier = if let Ok(user) = parts.extract_with_state::<User, _>(state).await {
user.id
} else {
to_real_ip(parts).await
};
let (bucket, resource) = storage.resolver.resolve_bucket(parts);
let limit = storage.resolver.resolve_bucket_limit(bucket);
let ratelimiter =
Ratelimiter::from(&storage.map, &identifier, limit, (bucket, resource));
parts.extensions.insert(ratelimiter.map_err(Json));
};
*parts
.extensions
.get::<Result<Ratelimiter, Json<Ratelimiter>>>()
.unwrap()
}
}
#[async_trait]
impl<S: Send + Sync> FromRequestParts<S> for RatelimitInformation
where
Database: FromRef<S>,
RatelimitStorage: FromRef<S>,
{
type Rejection = Json<RatelimitInformation>;
async fn from_request_parts(parts: &mut Parts, state: &S) -> Result<Self, Self::Rejection> {
if parts
.extensions
.get::<Result<Ratelimiter, Json<Ratelimiter>>>()
.is_none()
{
let ratelimiter = parts.extract_with_state::<Ratelimiter, S>(state).await;
parts.extensions.insert(ratelimiter);
};
let ratelimiter = *parts
.extensions
.get::<Result<Ratelimiter, Json<Ratelimiter>>>()
.unwrap();
match ratelimiter {
Ok(ratelimter) => Ok(RatelimitInformation::Success(ratelimter)),
Err(ratelimiter) => Err(Json(RatelimitInformation::Failure {
retry_after: ratelimiter.reset,
})),
}
}
}
pub async fn ratelimit_middleware(
State(database): State<Database>,
State(ratelimit_storage): State<RatelimitStorage>,
request: Request<Body>,
next: Next,
) -> Response {
#[derive(axum::extract::FromRef)]
struct TempState {
database: Database,
ratelimit_storage: RatelimitStorage,
}
let state = TempState {
database,
ratelimit_storage,
};
let (mut parts, body) = request.into_parts();
let res = Ratelimiter::from_request_parts(&mut parts, &state).await;
let (Ok(ratelimiter) | Err(Json(ratelimiter))) = &res;
let mut response = if res.is_ok() {
let request = Request::from_parts(parts, body);
next.run(request).await
} else {
let ratelimit_info = RatelimitInformation::from_request_parts(&mut parts, &state).await;
ratelimit_info.map(Json).into_response()
};
let Ratelimiter {
key,
limit,
remaining,
reset,
} = ratelimiter;
let headers = response.headers_mut();
headers.insert(
"X-RateLimit-Limit",
HeaderValue::from_str(&limit.to_string()).unwrap(),
);
headers.insert(
"X-RateLimit-Bucket",
HeaderValue::from_str(&key.to_string()).unwrap(),
);
headers.insert(
"X-RateLimit-Remaining",
HeaderValue::from_str(&remaining.to_string()).unwrap(),
);
headers.insert(
"X-RateLimit-Reset-After",
HeaderValue::from_str(&reset.to_string()).unwrap(),
);
if res.is_err() {
*response.status_mut() = StatusCode::TOO_MANY_REQUESTS;
};
response
}
async fn ratelimit_info(info: RatelimitInformation) -> Json<RatelimitInformation> {
Json(info)
}
pub fn routes<S: Clone + Send + Sync + 'static>() -> Router<S>
where
Database: FromRef<S>,
RatelimitStorage: FromRef<S>,
{
Router::new().route("/ratelimit", get(ratelimit_info))
}

View File

@@ -0,0 +1,7 @@
pub mod ratelimiter;
#[cfg(feature = "rocket")]
pub mod rocket;
#[cfg(feature = "axum")]
pub mod axum;

View File

@@ -0,0 +1,145 @@
use std::collections::hash_map::DefaultHasher;
use std::hash::Hasher;
use std::ops::Add;
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use serde::Serialize;
use dashmap::DashMap;
pub trait RequestKind {
type R<'a>;
}
pub trait RatelimitResolver<R>: Send + Sync {
fn resolve_bucket<'a>(&self, request: &'a R) -> (&'a str, Option<&'a str>);
fn resolve_bucket_limit(&self, bucket: &str) -> u32;
}
#[derive(Clone)]
pub struct RatelimitStorage<K: RequestKind> {
pub resolver: Arc<dyn for<'a> RatelimitResolver<K::R<'a>>>,
pub map: Arc<DashMap<u64, Entry>>,
}
impl<K: RequestKind> RatelimitStorage<K> {
pub fn new<R: for<'a> RatelimitResolver<K::R<'a>> + 'static>(resolver: R) -> Self {
Self {
resolver: Arc::new(resolver),
map: Arc::new(DashMap::new()),
}
}
}
/// Ratelimit Bucket
#[derive(Clone, Copy, Debug)]
pub struct Entry {
used: u32,
reset: u128,
}
/// Get the current time from Unix Epoch as a Duration
fn now() -> Duration {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.expect("Time went backwards...")
}
impl Entry {
/// Find bucket by its key
pub fn from(map: &DashMap<u64, Entry>, key: u64) -> Entry {
map.get(&key).map(|x| *x).unwrap_or_else(|| Entry {
used: 0,
reset: now().add(Duration::from_secs(10)).as_millis(),
})
}
/// Deduct one unit from the bucket and save
pub fn deduct(&mut self) {
let current_time = now().as_millis();
if current_time > self.reset {
self.used = 1;
self.reset = now().add(Duration::from_secs(10)).as_millis();
} else {
self.used += 1;
}
}
/// Save information
pub fn save(self, map: &DashMap<u64, Entry>, key: u64) {
map.insert(key, self);
}
/// Get remaining units in the bucket
pub fn get_remaining(&self, limit: u32) -> u32 {
if now().as_millis() > self.reset {
limit
} else {
limit - self.used
}
}
/// Get how long bucket has until reset
pub fn left_until_reset(&self) -> u128 {
let current_time = now().as_millis();
self.reset.saturating_sub(current_time)
}
}
/// Ratelimit Guard
#[derive(Serialize, Clone, Copy, Debug)]
#[allow(dead_code)]
pub struct Ratelimiter {
pub key: u64,
pub limit: u32,
pub remaining: u32,
pub reset: u128,
}
impl Ratelimiter {
/// Generate guard from identifier and target bucket
pub fn from(
map: &DashMap<u64, Entry>,
identifier: &str,
limit: u32,
(bucket, resource): (&str, Option<&str>),
) -> Result<Ratelimiter, Ratelimiter> {
let mut key = DefaultHasher::new();
key.write(identifier.as_bytes());
key.write(bucket.as_bytes());
if let Some(id) = resource {
key.write(id.as_bytes());
}
let key = key.finish();
let mut entry = Entry::from(map, key);
let remaining = entry.get_remaining(limit);
let reset = entry.left_until_reset();
let mut ratelimiter = Ratelimiter {
key,
limit,
remaining,
reset,
};
if remaining == 0 {
return Err(ratelimiter);
}
entry.deduct();
entry.save(map, key);
ratelimiter.remaining -= 1;
ratelimiter.reset = entry.left_until_reset();
Ok(ratelimiter)
}
}
#[derive(Serialize)]
#[serde(untagged)]
pub enum RatelimitInformation {
Success(Ratelimiter),
Failure { retry_after: u128 },
}

View File

@@ -0,0 +1,163 @@
use async_trait::async_trait;
use log::info;
use rocket::fairing::{Fairing, Info, Kind};
use rocket::http::uri::Origin;
use rocket::http::{Method, Status};
use rocket::request::{FromRequest, Outcome};
use rocket::serde::json::Json;
use rocket::{Data, Request, Response, State};
use revolt_config::config;
use revolt_rocket_okapi::r#gen::OpenApiGenerator;
use revolt_rocket_okapi::request::{OpenApiFromRequest, RequestHeaderInput};
use authifier::models::Session;
use crate::ratelimiter::RequestKind;
use crate::ratelimiter::{RatelimitInformation, Ratelimiter};
#[derive(Clone, Copy)]
pub struct RocketRequestKind;
impl RequestKind for RocketRequestKind {
type R<'a> = Request<'a>;
}
pub type RatelimitStorage = crate::ratelimiter::RatelimitStorage<RocketRequestKind>;
/// Find the remote IP of the client
fn to_ip(request: &'_ rocket::Request<'_>) -> String {
request
.remote()
.map(|x| x.ip().to_string())
.unwrap_or_default()
}
/// Find the actual IP of the client
async fn to_real_ip(request: &'_ rocket::Request<'_>) -> String {
if config().await.api.security.trust_cloudflare {
request
.headers()
.get_one("CF-Connecting-IP")
.map(|x| x.to_string())
.unwrap_or_else(|| to_ip(request))
} else {
to_ip(request)
}
}
#[async_trait]
impl<'r> FromRequest<'r> for Ratelimiter {
type Error = Ratelimiter;
async fn from_request<'a>(request: &'r rocket::Request<'a>) -> Outcome<Self, Self::Error> {
let ratelimiter = request
.local_cache_async(async {
use rocket::outcome::Outcome;
let storage = request.guard::<&State<RatelimitStorage>>().await.unwrap();
let identifier = if let Outcome::Success(session) = request.guard::<Session>().await
{
session.id
} else {
to_real_ip(request).await
};
let (bucket, resource) = storage.resolver.resolve_bucket(request);
let limit = storage.resolver.resolve_bucket_limit(bucket);
Ratelimiter::from(&storage.map, &identifier, limit, (bucket, resource))
})
.await;
match ratelimiter {
Ok(ratelimiter) => Outcome::Success(*ratelimiter),
Err(ratelimiter) => Outcome::Error((Status::TooManyRequests, *ratelimiter)),
}
}
}
impl OpenApiFromRequest<'_> for Ratelimiter {
fn from_request_input(
_gen: &mut OpenApiGenerator,
_name: String,
_required: bool,
) -> revolt_rocket_okapi::Result<RequestHeaderInput> {
Ok(RequestHeaderInput::None)
}
}
/// Attach ratelimiter to the Rocket application
pub struct RatelimitFairing;
#[async_trait]
impl Fairing for RatelimitFairing {
fn info(&self) -> Info {
Info {
name: "Ratelimiter",
kind: Kind::Request | Kind::Response,
}
}
async fn on_request(&self, request: &mut Request<'_>, _: &mut Data<'_>) {
use rocket::outcome::Outcome;
if let Outcome::Error(_) = request.guard::<Ratelimiter>().await {
info!(
"User rate-limited on route {}! (IP = {:?})",
request.uri(),
to_real_ip(request).await
);
request.set_method(Method::Get);
request.set_uri(Origin::parse("/ratelimit").unwrap())
}
}
async fn on_response<'r>(&self, request: &'r Request<'_>, response: &mut Response<'r>) {
let guard = request.guard::<Ratelimiter>().await;
let (Outcome::Success(ratelimiter) | Outcome::Error((_, ratelimiter))) = guard else {
unreachable!()
};
let Ratelimiter {
key,
limit,
remaining,
reset,
} = ratelimiter;
response.set_raw_header("X-RateLimit-Limit", limit.to_string());
response.set_raw_header("X-RateLimit-Bucket", key.to_string());
response.set_raw_header("X-RateLimit-Remaining", remaining.to_string());
response.set_raw_header("X-RateLimit-Reset-After", reset.to_string());
if guard.is_error() {
response.set_status(Status::TooManyRequests);
}
}
}
#[async_trait]
impl<'r> FromRequest<'r> for RatelimitInformation {
type Error = u128;
async fn from_request(request: &'r rocket::Request<'_>) -> Outcome<Self, Self::Error> {
let info = match request.guard::<Ratelimiter>().await {
Outcome::Success(ratelimiter) => RatelimitInformation::Success(ratelimiter),
Outcome::Error((_, ratelimiter)) => RatelimitInformation::Failure {
retry_after: ratelimiter.reset,
},
_ => unreachable!(),
};
Outcome::Success(info)
}
}
#[rocket::get("/ratelimit")]
fn ratelimit_info(info: RatelimitInformation) -> Json<RatelimitInformation> {
Json(info)
}
pub fn routes() -> Vec<rocket::Route> {
rocket::routes![ratelimit_info]
}

View File

@@ -1,6 +1,6 @@
[package]
name = "revolt-result"
version = "0.8.8"
version = "0.8.9"
edition = "2021"
license = "MIT"
authors = ["Paul Makles <me@insrt.uk>"]
@@ -15,7 +15,6 @@ utoipa = ["dep:utoipa"]
rocket = ["dep:rocket", "dep:serde_json"]
axum = ["dep:axum", "dep:serde_json"]
okapi = ["dep:revolt_rocket_okapi", "dep:revolt_okapi", "schemas"]
sentry = ["dep:sentry"]
default = ["serde"]
@@ -33,8 +32,7 @@ rocket = { optional = true, version = "0.5.0-rc.2", default-features = false }
revolt_rocket_okapi = { version = "0.10.0", optional = true }
revolt_okapi = { version = "0.9.1", optional = true }
# utilities
log = "0.4"
# Axum
axum = { version = "0.7.5", optional = true }
# Sentry
sentry = { version = "0.31.5", optional = true }

View File

@@ -2,9 +2,10 @@ use axum::{http::StatusCode, response::IntoResponse, Json};
use crate::{Error, ErrorType};
impl Error {
pub fn axum_status(&self) -> StatusCode {
match self.error_type {
/// HTTP response builder for Error enum
impl IntoResponse for Error {
fn into_response(self) -> axum::response::Response {
let status = match self.error_type {
ErrorType::LabelMe => StatusCode::INTERNAL_SERVER_ERROR,
ErrorType::AlreadyOnboarded => StatusCode::FORBIDDEN,
@@ -69,15 +70,16 @@ impl Error {
ErrorType::InvalidProperty => StatusCode::BAD_REQUEST,
ErrorType::InvalidSession => StatusCode::UNAUTHORIZED,
ErrorType::NotAuthenticated => StatusCode::UNAUTHORIZED,
ErrorType::Conflict => StatusCode::CONFLICT,
ErrorType::DuplicateNonce => StatusCode::CONFLICT,
ErrorType::VosoUnavailable => StatusCode::BAD_REQUEST,
ErrorType::NotFound => StatusCode::NOT_FOUND,
ErrorType::NoEffect => StatusCode::OK,
ErrorType::IOError => StatusCode::BAD_REQUEST,
ErrorType::UnprocessableEntity => StatusCode::UNPROCESSABLE_ENTITY,
ErrorType::DeserializationError { .. } => StatusCode::UNPROCESSABLE_ENTITY,
ErrorType::FailedValidation { .. } => StatusCode::UNPROCESSABLE_ENTITY,
ErrorType::FailedValidation { .. } => StatusCode::BAD_REQUEST,
ErrorType::LiveKitUnavailable => StatusCode::BAD_REQUEST,
ErrorType::NotConnected => StatusCode::BAD_REQUEST,
ErrorType::NotAVoiceChannel => StatusCode::BAD_REQUEST,
ErrorType::AlreadyConnected => StatusCode::BAD_REQUEST,
ErrorType::UnknownNode => StatusCode::BAD_REQUEST,
ErrorType::InvalidFlagValue => StatusCode::BAD_REQUEST,
ErrorType::FeatureDisabled { .. } => StatusCode::BAD_REQUEST,
@@ -87,13 +89,8 @@ impl Error {
ErrorType::FileTypeNotAllowed => StatusCode::BAD_REQUEST,
ErrorType::ImageProcessingFailed => StatusCode::INTERNAL_SERVER_ERROR,
ErrorType::NoEmbedData => StatusCode::BAD_REQUEST,
}
}
}
};
/// HTTP response builder for Error enum
impl IntoResponse for Error {
fn into_response(self) -> axum::response::Response {
(self.axum_status(), Json(&self)).into_response()
(status, Json(&self)).into_response()
}
}

View File

@@ -152,19 +152,19 @@ pub enum ErrorType {
InvalidSession,
InvalidFlagValue,
NotAuthenticated,
Conflict,
DuplicateNonce,
NotFound,
NoEffect,
IOError,
UnprocessableEntity,
DeserializationError {
error: String,
},
FailedValidation {
error: String,
},
// ? Voice errors
LiveKitUnavailable,
NotAVoiceChannel,
AlreadyConnected,
NotConnected,
UnknownNode,
// ? Micro-service errors
ProxyError,
FileTooSmall,
@@ -181,7 +181,7 @@ pub enum ErrorType {
// ? Feature flag disabled in the config
FeatureDisabled {
feature: String,
}
},
}
#[macro_export]
@@ -204,30 +204,37 @@ macro_rules! create_database_error {
};
}
pub trait ToRevoltError<T>: Sized {
fn capture_error(self) -> Self;
#[macro_export]
#[cfg(debug_assertions)]
macro_rules! query {
( $self: ident, $type: ident, $collection: expr, $($rest:expr),+ ) => {
Ok($self.$type($collection, $($rest),+).await.unwrap())
};
}
#[macro_export]
#[cfg(not(debug_assertions))]
macro_rules! query {
( $self: ident, $type: ident, $collection: expr, $($rest:expr),+ ) => {
$self.$type($collection, $($rest),+).await
.map_err(|_| create_database_error!(stringify!($type), $collection))
};
}
pub trait ToRevoltError<T> {
#[track_caller]
fn to_internal_error(self) -> Result<T, Error>;
}
impl<T, E: std::error::Error> ToRevoltError<T> for Result<T, E> {
fn capture_error(self) -> Self {
#[allow(unused_variables)]
self.inspect_err(|e| {
#[cfg(feature = "sentry")]
sentry::capture_error(e);
})
}
impl<T, E: std::fmt::Debug> ToRevoltError<T> for Result<T, E> {
#[track_caller]
fn to_internal_error(self) -> Result<T, Error> {
let loc = Location::caller();
self
.capture_error()
.map_err(|_| {
.map_err(|e| {
log::error!("{e:?}");
Error {
error_type: ErrorType::InternalError,
location: format!("{}:{}:{}", loc.file(), loc.line(), loc.column())
@@ -236,22 +243,12 @@ impl<T, E: std::error::Error> ToRevoltError<T> for Result<T, E> {
}
}
impl<T: std::error::Error> ToRevoltError<T> for Option<T> {
fn capture_error(self) -> Self {
#[allow(unused_variables)]
self.inspect(|e| {
#[cfg(feature = "sentry")]
sentry::capture_error(e);
})
}
impl<T> ToRevoltError<T> for Option<T> {
#[track_caller]
fn to_internal_error(self) -> Result<T, Error> {
let loc = Location::caller();
self
.capture_error()
.ok_or_else(|| {
self.ok_or_else(|| {
Error {
error_type: ErrorType::InternalError,
location: format!("{}:{}:{}", loc.file(), loc.line(), loc.column())

View File

@@ -8,9 +8,10 @@ use rocket::{
use crate::{Error, ErrorType};
impl Error {
pub fn rocket_status(&self) -> Status {
match self.error_type {
/// HTTP response builder for Error enum
impl<'r> Responder<'r, 'static> for Error {
fn respond_to(self, _: &'r Request<'_>) -> response::Result<'static> {
let status = match self.error_type {
ErrorType::LabelMe => Status::InternalServerError,
ErrorType::AlreadyOnboarded => Status::Forbidden,
@@ -76,15 +77,15 @@ impl Error {
ErrorType::InvalidProperty => Status::BadRequest,
ErrorType::InvalidSession => Status::Unauthorized,
ErrorType::NotAuthenticated => Status::Unauthorized,
ErrorType::Conflict => Status::Conflict,
ErrorType::DuplicateNonce => Status::Conflict,
ErrorType::VosoUnavailable => Status::BadRequest,
ErrorType::NotFound => Status::NotFound,
ErrorType::NoEffect => Status::Ok,
ErrorType::IOError => Status::BadRequest,
ErrorType::UnprocessableEntity => Status::UnprocessableEntity,
ErrorType::DeserializationError { .. } => Status::UnprocessableEntity,
ErrorType::FailedValidation { .. } => Status::UnprocessableEntity,
ErrorType::FailedValidation { .. } => Status::BadRequest,
ErrorType::LiveKitUnavailable => Status::BadRequest,
ErrorType::NotAVoiceChannel => Status::BadRequest,
ErrorType::AlreadyConnected => Status::BadRequest,
ErrorType::NotConnected => Status::BadRequest,
ErrorType::UnknownNode => Status::BadRequest,
ErrorType::FeatureDisabled { .. } => Status::BadRequest,
ErrorType::ProxyError => Status::BadRequest,
@@ -93,13 +94,9 @@ impl Error {
ErrorType::FileTypeNotAllowed => Status::BadRequest,
ErrorType::ImageProcessingFailed => Status::InternalServerError,
ErrorType::NoEmbedData => Status::BadRequest,
}
}
}
ErrorType::VosoUnavailable => Status::BadRequest,
};
/// HTTP response builder for Error enum
impl<'r> Responder<'r, 'static> for Error {
fn respond_to(self, _: &'r Request<'_>) -> response::Result<'static> {
// Serialize the error data structure into JSON.
let string = serde_json::to_string(&self).unwrap();
@@ -107,7 +104,7 @@ impl<'r> Responder<'r, 'static> for Error {
Response::build()
.sized_body(string.len(), Cursor::new(string))
.header(ContentType::new("application", "json"))
.status(self.rocket_status())
.status(status)
.ok()
}
}

View File

@@ -1,6 +1,6 @@
[package]
name = "revolt-crond"
version = "0.8.8"
version = "0.8.9"
license = "AGPL-3.0-or-later"
authors = ["Paul Makles <me@insrt.uk>"]
edition = "2021"
@@ -16,7 +16,7 @@ log = "0.4"
tokio = { version = "1" }
# Core
revolt-database = { version = "0.8.8", path = "../../core/database" }
revolt-result = { version = "0.8.8", path = "../../core/result" }
revolt-config = { version = "0.8.8", path = "../../core/config" }
revolt-files = { version = "0.8.8", path = "../../core/files" }
revolt-database = { version = "0.8.9", path = "../../core/database" }
revolt-result = { version = "0.8.9", path = "../../core/result" }
revolt-config = { version = "0.8.9", path = "../../core/config" }
revolt-files = { version = "0.8.9", path = "../../core/files" }

View File

@@ -1,5 +1,5 @@
# Build Stage
FROM ghcr.io/revoltchat/base:latest AS builder
FROM ghcr.io/stoatchat/base:latest AS builder
FROM debian:12 AS debian
# Bundle Stage

View File

@@ -1,32 +1,20 @@
use std::{future::Future, time::Duration};
use revolt_config::{configure, capture_error};
use revolt_database::{Database, DatabaseInfo};
use revolt_config::configure;
use revolt_database::DatabaseInfo;
use revolt_result::Result;
use tasks::{file_deletion, prune_dangling_files};
use tokio::{join, time::sleep};
use tasks::{file_deletion, prune_dangling_files, prune_members};
use tokio::try_join;
pub mod tasks;
pub async fn cron_task_wrapper<Fut: Future<Output = Result<()>>>(func: fn(Database) -> Fut, db: Database) {
loop {
if let Err(error) = func(db.clone()).await {
log::error!("cron task failed unexpectidly: {error:?}\nRetrying after 60s");
capture_error(&error);
}
sleep(Duration::from_secs(60)).await;
}
}
#[tokio::main]
async fn main() {
async fn main() -> Result<()> {
configure!(crond);
let db = DatabaseInfo::Auto.connect().await.expect("database");
join!(
cron_task_wrapper(file_deletion::task, db.clone()),
cron_task_wrapper(prune_dangling_files::task, db.clone()),
);
try_join!(
file_deletion::task(db.clone()),
prune_dangling_files::task(db.clone()),
prune_members::task(db.clone())
)
.map(|_| ())
}

View File

@@ -1,2 +1,3 @@
pub mod file_deletion;
pub mod prune_dangling_files;
pub mod prune_members;

View File

@@ -0,0 +1,18 @@
use std::time::Duration;
use log::warn;
use revolt_database::Database;
use revolt_result::Result;
use tokio::time::sleep;
pub async fn task(db: Database) -> Result<()> {
loop {
let success = db.remove_dangling_members().await;
if let Err(s) = success {
revolt_config::capture_error(&s);
warn!("Failed to prune dangling members: {:?}", &s);
}
sleep(Duration::from_secs(90)).await;
}
}

View File

@@ -1,20 +1,20 @@
[package]
name = "revolt-pushd"
version = "0.8.8"
version = "0.8.9"
edition = "2021"
license = "AGPL-3.0-or-later"
[dependencies]
revolt-result = { version = "0.8.8", path = "../../core/result" }
revolt-config = { version = "0.8.8", path = "../../core/config", features = [
revolt-result = { version = "0.8.9", path = "../../core/result" }
revolt-config = { version = "0.8.9", path = "../../core/config", features = [
"report-macros",
"anyhow"
] }
revolt-database = { version = "0.8.8", path = "../../core/database" }
revolt-models = { version = "0.8.8", path = "../../core/models", features = [
revolt-database = { version = "0.8.9", path = "../../core/database" }
revolt-models = { version = "0.8.9", path = "../../core/models", features = [
"validator",
] }
revolt-presence = { version = "0.8.8", path = "../../core/presence", features = [
revolt-presence = { version = "0.8.9", path = "../../core/presence", features = [
"redis-is-patched",
] }

View File

@@ -1,5 +1,5 @@
# Build Stage
FROM ghcr.io/revoltchat/base:latest AS builder
FROM ghcr.io/stoatchat/base:latest AS builder
FROM debian:12 AS debian
# Bundle Stage

View File

@@ -123,26 +123,24 @@ impl AsyncConsumer for AckConsumer {
token: session.subscription.as_ref().unwrap().auth.clone(),
extras: Default::default(),
};
let raw_service_payload = serde_json::to_string(&service_payload);
match serde_json::to_string(&service_payload) {
Ok(p) => {
let args = BasicPublishArguments::new(
config.pushd.exchange.as_str(),
config.pushd.apn.queue.as_str(),
)
.finish();
if let Ok(p) = raw_service_payload {
let args = BasicPublishArguments::new(
config.pushd.exchange.as_str(),
config.pushd.apn.queue.as_str(),
)
.finish();
log::debug!(
"Publishing ack to apn session {}",
session.subscription.as_ref().unwrap().auth
);
log::debug!(
"Publishing ack to apn session {}",
session.subscription.as_ref().unwrap().auth
);
publish_message(self, p.into(), args).await;
},
Err(e) => {
log::warn!("Failed to serialize ack badge update payload!");
revolt_config::capture_error(&e);
}
publish_message(self, p.into(), args).await;
} else {
log::warn!("Failed to serialize ack badge update payload!");
revolt_config::capture_error(&raw_service_payload.unwrap_err());
}
}
}

View File

@@ -0,0 +1,168 @@
use std::collections::HashMap;
use crate::consumers::inbound::internal::*;
use amqprs::{
channel::{BasicPublishArguments, Channel},
connection::Connection,
consumer::AsyncConsumer,
BasicProperties, Deliver,
};
use anyhow::Result;
use async_trait::async_trait;
use log::debug;
use revolt_database::{events::rabbit::*, Database};
pub struct DmCallConsumer {
#[allow(dead_code)]
db: Database,
authifier_db: authifier::Database,
conn: Option<Connection>,
channel: Option<Channel>,
}
impl Channeled for DmCallConsumer {
fn get_connection(&self) -> Option<&Connection> {
if self.conn.is_none() {
None
} else {
Some(self.conn.as_ref().unwrap())
}
}
fn get_channel(&self) -> Option<&Channel> {
if self.channel.is_none() {
None
} else {
Some(self.channel.as_ref().unwrap())
}
}
fn set_connection(&mut self, conn: Connection) {
self.conn = Some(conn);
}
fn set_channel(&mut self, channel: Channel) {
self.channel = Some(channel)
}
}
impl DmCallConsumer {
pub fn new(db: Database, authifier_db: authifier::Database) -> DmCallConsumer {
DmCallConsumer {
db,
authifier_db,
conn: None,
channel: None,
}
}
async fn consume_event(
&mut self,
_channel: &Channel,
_deliver: Deliver,
_basic_properties: BasicProperties,
content: Vec<u8>,
) -> Result<()> {
let content = String::from_utf8(content)?;
let _p: InternalDmCallPayload = serde_json::from_str(content.as_str())?;
let payload = _p.payload;
debug!("Received dm call start/stop event");
let (revolt_database::Channel::DirectMessage { recipients, .. }
| revolt_database::Channel::Group { recipients, .. }) =
self.db.fetch_channel(&payload.channel_id).await?
else {
warn!(
"Discarding dm call start/stop event for non-dm/group channel {}",
payload.channel_id
);
return Ok(());
};
let call_recipients = if let Some(user_recipients) = _p.recipients {
user_recipients
.into_iter()
.filter(|user_id| recipients.contains(user_id) && user_id != &payload.initiator_id)
.collect()
} else {
recipients
.into_iter()
.filter(|user_id| user_id != &payload.initiator_id)
.collect::<Vec<_>>()
};
let config = revolt_config::config().await;
for user_id in call_recipients {
if let Ok(sessions) = self.authifier_db.find_sessions(&user_id).await {
for session in sessions {
if let Some(sub) = session.subscription {
let mut sendable = PayloadToService {
notification: PayloadKind::DmCallStartEnd(payload.clone()),
token: sub.auth,
user_id: session.user_id,
session_id: session.id,
extras: HashMap::new(),
};
let args: BasicPublishArguments;
if sub.endpoint == "apn" {
args = BasicPublishArguments::new(
config.pushd.exchange.as_str(),
config.pushd.apn.queue.as_str(),
)
.finish();
} else if sub.endpoint == "fcm" {
args = BasicPublishArguments::new(
config.pushd.exchange.as_str(),
config.pushd.fcm.queue.as_str(),
)
.finish();
} else {
// web push (vapid)
args = BasicPublishArguments::new(
config.pushd.exchange.as_str(),
config.pushd.vapid.queue.as_str(),
)
.finish();
sendable.extras.insert("p265dh".to_string(), sub.p256dh);
sendable
.extras
.insert("endpoint".to_string(), sub.endpoint.clone());
}
let payload = serde_json::to_string(&sendable)?;
publish_message(self, payload.into(), args).await;
}
}
}
}
Ok(())
}
}
#[allow(unused_variables)]
#[async_trait]
impl AsyncConsumer for DmCallConsumer {
/// This consumer handles delegating messages into their respective platform queues.
async fn consume(
&mut self,
channel: &Channel,
deliver: Deliver,
basic_properties: BasicProperties,
content: Vec<u8>,
) {
if let Err(err) = self
.consume_event(channel, deliver, basic_properties, content)
.await
{
revolt_config::capture_anyhow(&err);
warn!("Failed to process dm call start/stop event: {err:?}");
}
}
}

Some files were not shown because too many files have changed in this diff Show More