-
-
+
+
+
+
+
+
+ {{ callPeople }} in it.
+ Join with audio
+ Join call
+
+
+
diff --git a/apps/web/vite.config.ts b/apps/web/vite.config.ts
index af84200..4ed93fa 100644
--- a/apps/web/vite.config.ts
+++ b/apps/web/vite.config.ts
@@ -26,6 +26,8 @@ export default defineConfig({
proxy: {
'/_matrix/client': { target: homeserver, changeOrigin: true },
'/_matrix/media': { target: homeserver, changeOrigin: true },
+ // Where the client finds the group call backend (rtc_foci).
+ '/.well-known/matrix/client': { target: homeserver, changeOrigin: true },
},
},
})
diff --git a/deploy/compose/calls.yml b/deploy/compose/calls.yml
new file mode 100644
index 0000000..a5215be
--- /dev/null
+++ b/deploy/compose/calls.yml
@@ -0,0 +1,117 @@
+# Calls for a local Warp: a LiveKit SFU and the LiveKit JWT service
+# (lk-jwt-service), layered over docker-compose.yml.
+#
+# cd deploy/compose
+# docker compose -f docker-compose.yml -f calls.yml --profile web up -d --build
+#
+# Then open http://localhost:8080 in a few browsers (or private windows),
+# sign in as different users, and press the voice or video call button in
+# a room: in a room of two it rings the other person.
+#
+# How the pieces find each other:
+#
+# browser ── /.well-known/matrix/client ──▶ warp rtc_foci: http://localhost:8070
+# browser ── openid/request_token ────────▶ warp an OpenID token
+# browser ── POST /sfu/get ───────────────▶ jwt {url, jwt} for the LiveKit room
+# jwt ───── /_matrix/federation/v1/openid/userinfo ─▶ warp (https://:8448)
+# browser ── ws://localhost:7880 + UDP 7882 ─▶ livekit
+#
+# The JWT service checks a token by asking the user's homeserver over the
+# federation port, with TLS. To make that work with the default server name
+# "localhost", warp gets a federation listener on :8448 with a self-signed
+# certificate, and the JWT service and LiveKit share warp's network
+# namespace, so that "localhost" means warp for all three. This is a local
+# setup: in production, give each its own hostname and real certificates
+# (see docs/calls.md).
+
+services:
+ # Makes the self-signed certificate for warp's federation listener once.
+ calls-cert:
+ image: alpine:3.21
+ restart: "no"
+ entrypoint: ["/bin/sh", "-c"]
+ command:
+ - >-
+ test -f /tls/tls.crt ||
+ (apk add --no-cache openssl >/dev/null &&
+ openssl req -x509 -newkey rsa:2048 -nodes -days 3650
+ -subj "/CN=$${WARP_SERVER_NAME}" -addext "subjectAltName=DNS:$${WARP_SERVER_NAME}"
+ -keyout /tls/tls.key -out /tls/tls.crt &&
+ chmod 644 /tls/tls.key /tls/tls.crt)
+ environment:
+ WARP_SERVER_NAME: ${WARP_SERVER_NAME:-localhost}
+ volumes:
+ - warp-tls:/tls
+
+ warp:
+ depends_on:
+ calls-cert:
+ condition: service_completed_successfully
+ environment:
+ WARP_RTC_LIVEKIT_SERVICE_URL: ${WARP_LIVEKIT_JWT_PUBLIC_URL:-http://localhost:8070}
+ WARP_LISTEN_FEDERATION_ADDRESS: ":8448"
+ WARP_TLS_CERT_FILE: /tls/tls.crt
+ WARP_TLS_KEY_FILE: /tls/tls.key
+ volumes:
+ - warp-tls:/tls:ro
+ # LiveKit and the JWT service live in this container's network
+ # namespace, so their ports are published here.
+ ports:
+ - "8070:8070" # lk-jwt-service
+ - "7880:7880" # LiveKit signalling (WebSocket)
+ - "7881:7881" # LiveKit media over TCP, the fallback
+ - "7882:7882/udp" # LiveKit media over UDP
+
+ livekit:
+ image: livekit/livekit-server:v1.13.7
+ restart: unless-stopped
+ network_mode: service:warp
+ depends_on:
+ - warp
+ environment:
+ # node_ip is what LiveKit tells browsers to send media to. 127.0.0.1
+ # works for browsers on this machine; set LIVEKIT_NODE_IP to the
+ # machine's LAN address to call from other devices.
+ LIVEKIT_CONFIG: |
+ port: 7880
+ bind_addresses: [""]
+ rtc:
+ tcp_port: 7881
+ udp_port: 7882
+ use_external_ip: false
+ node_ip: ${LIVEKIT_NODE_IP:-127.0.0.1}
+ room:
+ auto_create: false
+ keys:
+ ${LIVEKIT_KEY:-warp}: ${LIVEKIT_SECRET:-warp-local-livekit-secret-change-me-0123456789}
+ logging:
+ level: info
+
+ livekit-jwt:
+ image: ghcr.io/element-hq/lk-jwt-service:0.8.0
+ restart: unless-stopped
+ network_mode: service:warp
+ depends_on:
+ - warp
+ - livekit
+ environment:
+ LIVEKIT_JWT_BIND: ":8070"
+ # The URL handed to browsers, and the one the service creates rooms
+ # through: both reach LiveKit at localhost here.
+ LIVEKIT_URL: ${LIVEKIT_PUBLIC_URL:-ws://localhost:7880}
+ LIVEKIT_KEY: ${LIVEKIT_KEY:-warp}
+ LIVEKIT_SECRET: ${LIVEKIT_SECRET:-warp-local-livekit-secret-change-me-0123456789}
+ # Users of this server may create call rooms on the SFU.
+ LIVEKIT_FULL_ACCESS_HOMESERVERS: ${WARP_SERVER_NAME:-localhost}
+ # warp's federation certificate is self-signed.
+ LIVEKIT_INSECURE_SKIP_VERIFY_TLS: YES_I_KNOW_WHAT_I_AM_DOING
+
+ web:
+ environment:
+ # The browser connects to the JWT service and to LiveKit directly.
+ WARP_WEB_CALL_ORIGINS: >-
+ ${WARP_LIVEKIT_JWT_PUBLIC_URL:-http://localhost:8070}
+ ${LIVEKIT_PUBLIC_URL:-ws://localhost:7880}
+
+volumes:
+ warp-tls:
diff --git a/deploy/compose/docker-compose.yml b/deploy/compose/docker-compose.yml
index 966aebe..0dec621 100644
--- a/deploy/compose/docker-compose.yml
+++ b/deploy/compose/docker-compose.yml
@@ -12,6 +12,12 @@
# http://localhost:8009; with the web profile, the web client is on
# http://localhost:8080.
#
+# For voice and video calls (LiveKit and its JWT service) add calls.yml:
+#
+# docker compose -f docker-compose.yml -f calls.yml --profile web up -d
+#
+# WARP_PORT moves the homeserver off host port 8008.
+#
# Create the first administrator once the server is up:
#
# docker compose exec warp warp create-account --username admin --admin
@@ -63,7 +69,7 @@ services:
volumes:
- warp-data:/data
ports:
- - "8008:8008"
+ - "${WARP_PORT:-8008}:8008"
# The admin panel is optional. Start it with the admin profile:
# docker compose --profile admin up -d
diff --git a/deploy/helm/warp/templates/_helpers.tpl b/deploy/helm/warp/templates/_helpers.tpl
index 8be9db1..77b0055 100644
--- a/deploy/helm/warp/templates/_helpers.tpl
+++ b/deploy/helm/warp/templates/_helpers.tpl
@@ -65,3 +65,32 @@ app.kubernetes.io/name: {{ printf "%s-admin" (include "warp.name" .) }}
app.kubernetes.io/instance: {{ .Release.Name }}
app.kubernetes.io/component: admin
{{- end -}}
+
+{{/*
+The LiveKit JWT service for calls, with its own name label like the
+admin panel.
+*/}}
+{{- define "warp.calls.fullname" -}}
+{{- printf "%s-livekit-jwt" (include "warp.fullname" . | trunc 51 | trimSuffix "-") -}}
+{{- end -}}
+
+{{- define "warp.calls.labels" -}}
+helm.sh/chart: {{ printf "%s-%s" .Chart.Name .Chart.Version | replace "+" "_" }}
+{{ include "warp.calls.selectorLabels" . }}
+app.kubernetes.io/managed-by: {{ .Release.Service }}
+{{- end -}}
+
+{{- define "warp.calls.selectorLabels" -}}
+app.kubernetes.io/name: {{ printf "%s-livekit-jwt" (include "warp.name" .) }}
+app.kubernetes.io/instance: {{ .Release.Name }}
+app.kubernetes.io/component: calls
+{{- end -}}
+
+{{/* The JWT service's public URL, advertised to clients: explicit, else its ingress. */}}
+{{- define "warp.calls.serviceURL" -}}
+{{- if .Values.calls.livekitServiceURL -}}
+{{- .Values.calls.livekitServiceURL -}}
+{{- else if and .Values.calls.jwtService.enabled .Values.calls.jwtService.ingress.enabled -}}
+{{- printf "https://%s" .Values.calls.jwtService.ingress.host -}}
+{{- end -}}
+{{- end -}}
diff --git a/deploy/helm/warp/templates/deployment.yaml b/deploy/helm/warp/templates/deployment.yaml
index 2cef553..82b6e75 100644
--- a/deploy/helm/warp/templates/deployment.yaml
+++ b/deploy/helm/warp/templates/deployment.yaml
@@ -97,6 +97,10 @@ spec:
value: {{ .Values.tracing.sampleRatio | quote }}
- name: WARP_LOG_FORMAT
value: json
+ {{- with (include "warp.calls.serviceURL" .) }}
+ - name: WARP_RTC_LIVEKIT_SERVICE_URL
+ value: {{ . | quote }}
+ {{- end }}
{{- with .Values.extraEnv }}
{{- toYaml . | nindent 12 }}
{{- end }}
diff --git a/deploy/helm/warp/templates/livekit-jwt.yaml b/deploy/helm/warp/templates/livekit-jwt.yaml
new file mode 100644
index 0000000..5717e15
--- /dev/null
+++ b/deploy/helm/warp/templates/livekit-jwt.yaml
@@ -0,0 +1,107 @@
+{{- if .Values.calls.jwtService.enabled }}
+{{- $jwt := .Values.calls.jwtService }}
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+ name: {{ include "warp.calls.fullname" . }}
+ labels:
+ {{- include "warp.calls.labels" . | nindent 4 }}
+spec:
+ # The service keeps its state in memory: one replica, unless LIVEKIT_REDIS_URL
+ # is set through extraEnv.
+ replicas: 1
+ selector:
+ matchLabels:
+ {{- include "warp.calls.selectorLabels" . | nindent 6 }}
+ template:
+ metadata:
+ labels:
+ {{- include "warp.calls.selectorLabels" . | nindent 8 }}
+ spec:
+ {{- with .Values.imagePullSecrets }}
+ imagePullSecrets:
+ {{- toYaml . | nindent 8 }}
+ {{- end }}
+ containers:
+ - name: lk-jwt-service
+ image: "{{ $jwt.image.repository }}:{{ $jwt.image.tag }}"
+ imagePullPolicy: {{ $jwt.image.pullPolicy }}
+ ports:
+ - name: http
+ containerPort: 8080
+ protocol: TCP
+ env:
+ - name: LIVEKIT_JWT_BIND
+ value: ":8080"
+ - name: LIVEKIT_URL
+ value: {{ required "calls.jwtService.livekitURL is required" $jwt.livekitURL | quote }}
+ - name: LIVEKIT_KEY
+ valueFrom:
+ secretKeyRef:
+ name: {{ required "calls.jwtService.existingSecret is required" $jwt.existingSecret }}
+ key: {{ $jwt.keyKey }}
+ - name: LIVEKIT_SECRET
+ valueFrom:
+ secretKeyRef:
+ name: {{ $jwt.existingSecret }}
+ key: {{ $jwt.secretKey }}
+ - name: LIVEKIT_FULL_ACCESS_HOMESERVERS
+ value: {{ $jwt.fullAccessHomeservers | default .Values.serverName | quote }}
+ {{- with $jwt.extraEnv }}
+ {{- toYaml . | nindent 12 }}
+ {{- end }}
+ readinessProbe:
+ httpGet:
+ path: /healthz
+ port: http
+ resources:
+ {{- toYaml $jwt.resources | nindent 12 }}
+---
+apiVersion: v1
+kind: Service
+metadata:
+ name: {{ include "warp.calls.fullname" . }}
+ labels:
+ {{- include "warp.calls.labels" . | nindent 4 }}
+spec:
+ type: {{ $jwt.service.type }}
+ ports:
+ - name: http
+ port: {{ $jwt.service.port }}
+ targetPort: http
+ protocol: TCP
+ selector:
+ {{- include "warp.calls.selectorLabels" . | nindent 4 }}
+{{- if $jwt.ingress.enabled }}
+---
+apiVersion: networking.k8s.io/v1
+kind: Ingress
+metadata:
+ name: {{ include "warp.calls.fullname" . }}
+ labels:
+ {{- include "warp.calls.labels" . | nindent 4 }}
+ {{- with $jwt.ingress.annotations }}
+ annotations:
+ {{- toYaml . | nindent 4 }}
+ {{- end }}
+spec:
+ {{- with $jwt.ingress.className }}
+ ingressClassName: {{ . }}
+ {{- end }}
+ {{- with $jwt.ingress.tls }}
+ tls:
+ {{- toYaml . | nindent 4 }}
+ {{- end }}
+ rules:
+ - host: {{ required "calls.jwtService.ingress.host is required" $jwt.ingress.host | quote }}
+ http:
+ paths:
+ - path: /
+ pathType: Prefix
+ backend:
+ service:
+ name: {{ include "warp.calls.fullname" . }}
+ port:
+ name: http
+{{- end }}
+{{- end }}
diff --git a/deploy/helm/warp/values.yaml b/deploy/helm/warp/values.yaml
index 318f0e0..e4c3c24 100644
--- a/deploy/helm/warp/values.yaml
+++ b/deploy/helm/warp/values.yaml
@@ -190,3 +190,50 @@ admin:
nodeSelector: {}
tolerations: []
affinity: {}
+
+# Calls (MatrixRTC with a LiveKit SFU). Warp only advertises the LiveKit
+# JWT service and answers its OpenID checks; the media goes through LiveKit.
+# See docs/calls.md.
+calls:
+ # The JWT service's public URL, advertised in /.well-known/matrix/client.
+ # Defaults to https:// when the chart runs the
+ # service with an ingress; empty and without one, there are no calls.
+ livekitServiceURL: ""
+
+ # lk-jwt-service: hands out LiveKit tokens to users of this server, who
+ # prove who they are with an OpenID token. It checks that token with this
+ # server's federation endpoint at https:///.well-known/matrix/server
+ # or :8448, so one of the two must reach Warp.
+ jwtService:
+ enabled: false
+ image:
+ repository: ghcr.io/element-hq/lk-jwt-service
+ tag: "0.8.0"
+ pullPolicy: IfNotPresent
+ # The LiveKit SFU's WebSocket URL as browsers reach it, e.g.
+ # wss://livekit.example.org. Run LiveKit itself with its own chart
+ # (https://github.com/livekit/livekit-helm); it needs UDP from clients.
+ livekitURL: wss://livekit.example.org
+ # A secret with the LiveKit API key and secret under these keys.
+ existingSecret: ""
+ keyKey: LIVEKIT_KEY
+ secretKey: LIVEKIT_SECRET
+ # Homeservers whose users may create call rooms on the SFU. Defaults to
+ # serverName.
+ fullAccessHomeservers: ""
+ service:
+ type: ClusterIP
+ port: 8080
+ ingress:
+ enabled: false
+ className: ""
+ annotations: {}
+ host: livekit-jwt.example.org
+ tls: []
+ extraEnv: []
+ resources:
+ requests:
+ cpu: 10m
+ memory: 32Mi
+ limits:
+ memory: 128Mi
diff --git a/docs/README.md b/docs/README.md
index 12b584b..b41f95c 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -13,6 +13,7 @@
| [federation.md](federation.md) | the Server-Server API: discovery, keys, taking in events, joins across servers, the outbound queue, the switch |
| [media.md](media.md) | `internal/mediaapi`: uploads, downloads, thumbnails, other servers' media, where the files go |
| [e2ee.md](e2ee.md) | `internal/keyserver`: device and one-time keys, cross-signing, device list tracking, key backup |
+| [calls.md](calls.md) | voice and video calls with MatrixRTC and LiveKit: ringing, delayed events, OpenID, the JWT service, deploying it |
| [push.md](push.md) | `internal/pushrules` and `internal/pushserver`: push rules, notification counts, pushers, the push gateway client |
| [appservices.md](appservices.md) | `internal/appservice`: registration files, namespaces, acting as users, delivering events to bridges |
| [admin-api.md](admin-api.md) | `/_warp/admin/v1`: the operator API behind the admin panel |
diff --git a/docs/calls.md b/docs/calls.md
new file mode 100644
index 0000000..0b48ec4
--- /dev/null
+++ b/docs/calls.md
@@ -0,0 +1,171 @@
+# Voice and video calls
+
+Every call in Warp's web client, between two people or twenty, is a
+MatrixRTC call (MSC4143) with its media through a LiveKit SFU: the Element
+Call model, which Element X uses too. A call of two is a call with two
+members, plus a ring. The server needs a LiveKit SFU and the LiveKit JWT
+service (`client_api.rtc`); without them there are no calls.
+
+The old Matrix 1:1 VoIP (`m.call.*` events, WebRTC between the two
+browsers) is not used: the web client starts the SDK with `disableVoip`, so
+it neither places nor answers such calls, and shows them only as history in
+the timeline. The server still serves `GET /_matrix/client/v3/voip/turnServer`
+(`client_api.turn`) for other clients that do.
+
+## How a call works
+
+Who is in a room's call is room state: every device in it has an
+`org.matrix.msc3401.call.member` state event (state key
+`_@user:server_DEVICE_m.call`), with the LiveKit service it publishes to.
+The media goes through a LiveKit SFU, which Warp does not run itself.
+
+```
+browser ── GET /.well-known/matrix/client ──────────▶ warp org.matrix.msc4143.rtc_foci
+ ── GET …/org.matrix.msc4143/rtc/transports ─▶ warp (the same list, authenticated)
+ ── PUT …/state/org.matrix.msc3401.call.member/
+ ?org.matrix.msc4140.delay=… (the delayed "left")
+ and then without a delay (the "joined") ──▶ warp
+ ── POST /user//openid/request_token ────▶ warp an OpenID token
+ ── POST /sfu/get ──────▶ lk-jwt-service
+ lk-jwt-service ── GET /_matrix/federation/v1/openid/userinfo ──▶ warp
+ ◀── {url, jwt}
+ ── WebSocket + UDP ─────────────────────────▶ LiveKit
+```
+
+### What the server does
+
+- **Discovery.** With `client_api.rtc.livekit_service_url` set, Warp adds
+ `org.matrix.msc4143.rtc_foci: [{"type": "livekit", "livekit_service_url": …}]`
+ to `/.well-known/matrix/client` and serves the same list at
+ `GET /_matrix/client/unstable/org.matrix.msc4143/rtc/transports`.
+- **Delayed events (MSC4140)**, advertised as `org.matrix.msc4140` in
+ `/versions`. `PUT /rooms/{roomId}/state/{type}/{key}` and
+ `PUT /rooms/{roomId}/send/{type}/{txnId}` take
+ `?org.matrix.msc4140.delay=`: the event is stored instead of sent,
+ and the answer is `{"delay_id": …}`. Under
+ `/_matrix/client/unstable/org.matrix.msc4140`:
+ - `GET /delayed_events` lists your events as `scheduled` and `finalised`
+ (`?status=` and `?delay_id=` narrow it); `GET /delayed_events/{id}`
+ returns one.
+ - `POST /delayed_events/{id}/{restart|cancel|send}`, or the older
+ `POST /delayed_events/{id}` with `{"action": …}`. These need no access
+ token: the delay ID is the capability.
+
+ A scheduler goroutine sleeps until the next event is due and sends it
+ as the user who scheduled it, through the same pipeline as any event, so
+ power levels and membership still apply; a send that fails is finalised
+ with the error the client would have got. Someone *else* setting the
+ same piece of state cancels a delayed state event, so a stale "left"
+ can't overwrite a newer membership; the owner's own updates don't. Events
+ live in the database (`delayed_events`, migration 14) and survive a
+ restart. Finalised events are kept for a day. `client_api.delayed_events`
+ caps the delay (24h) and how many a user may have (100).
+
+ This is what makes a call robust: a joining member schedules its own
+ empty `call.member` a few seconds out and keeps restarting it while it
+ is in the call. A browser that crashes or loses its network stops
+ restarting it, and Warp removes the member.
+- **OpenID.** `POST /_matrix/client/v3/user/{userId}/openid/request_token`
+ gives a user a one-hour token, which `GET /_matrix/federation/v1/openid/userinfo?access_token=…`
+ turns back into `{"sub": "@user:server"}` for whoever it was handed to.
+ Tokens are stored hashed (`openid_tokens`). The userinfo endpoint is
+ served with federation switched off too: a server that does not
+ federate can still have calls.
+- **Power levels.** New rooms let anyone who may talk join the call:
+ `org.matrix.msc3401.call.member` is at level 0 in the `events` of the
+ default power levels. Rooms created before this keep `state_default`
+ (50) for it, so an ordinary member gets `M_FORBIDDEN` until a moderator
+ sets it to 0.
+
+### Ringing
+
+Starting a call in a room with one other person rings them. The caller's
+MatrixRTC session sends an `org.matrix.msc4075.rtc.notification`
+(`notification_type: "ring"`, a `lifetime` of a minute, `m.call.intent`
+audio or video) when its membership is the first; in a bigger room it
+sends `notification`, which the web client doesn't ring for, since the
+"Join call" banner says enough. The other side shows an incoming-call card
+for a ring that is someone else's, mentions them or the room, and is
+within its lifetime, until they answer (join the call), decline (an
+`org.matrix.msc4310.rtc.decline` referencing the notification, which also
+stops the ring on their other devices), the caller leaves, or it expires.
+The caller hears a ringback until someone joins, and gives up on a decline
+or after the minute. In an encrypted room both events are encrypted like
+any other.
+
+The SDK's own validator for received notifications expects MSC4354 sticky
+events and `m.rtc.slot` state, which Warp doesn't serve yet, so the web
+client applies the receiving rules itself.
+
+### What the web client does
+
+`src/stores/call.ts` uses the SDK's MatrixRTC session
+(`client.matrixRTC.getRoomSession(room)`), publishing the transport
+`{type: "livekit", livekit_service_url, livekit_alias: }` (the one
+the call already uses, else this server's), gets a LiveKit token from
+`/sfu/get`, and connects with `livekit-client`. In an encrypted room it
+joins with `manageMediaKeys`: each member makes its own media key and
+sends it to the others over encrypted to-device messages, and a
+`BaseKeyProvider` hands each key to LiveKit's frame encryption for that
+participant (`@user:server:DEVICE`). The settings (no shared key, no
+ratcheting) match Element Call's, so the two can share a call.
+
+### Running it
+
+Locally, [`deploy/compose/calls.yml`](../deploy/compose/calls.yml) adds
+LiveKit and lk-jwt-service to the Compose setup:
+
+```sh
+cd deploy/compose
+WARP_REGISTRATION_ENABLED=true docker compose -f docker-compose.yml -f calls.yml --profile web up -d --build
+```
+
+Then sign in as different users at ,
+ and (different origins keep
+separate sessions in one browser). A voice or video call in a room of two
+rings the other person; in a room of three or more the others see a "Join
+call" banner. `WARP_PORT` and `WARP_WEB_PORT` move the homeserver and web client
+off 8008 and 8080 (then set `WARP_PUBLIC_BASEURL` to match).
+
+The JWT service checks an OpenID token by asking the user's homeserver
+over its federation port with TLS. For the server name `localhost` that
+means `https://localhost:8448`, so the file gives Warp a federation
+listener with a self-signed certificate (`LIVEKIT_INSECURE_SKIP_VERIFY_TLS`
+accepts it) and runs LiveKit and the JWT service in Warp's network
+namespace, where `localhost` is Warp. One consequence: when the `warp`
+container is recreated on its own, the other two lose their network. Run
+`up` for the whole file, not for one service, or recreate them with
+`docker compose -f docker-compose.yml -f calls.yml up -d --force-recreate livekit livekit-jwt`.
+LiveKit advertises `127.0.0.1` for media; set `LIVEKIT_NODE_IP` to the
+machine's LAN address to call from other devices.
+
+In production, give each piece its own hostname and a real certificate:
+
+- **LiveKit**, reachable by browsers on TCP 7880 behind TLS (`wss://`),
+ and UDP (7882, or a port range) and TCP 7881 for media. See LiveKit's
+ own deployment guide or its Helm chart.
+- **lk-jwt-service** behind TLS, with `LIVEKIT_URL` (the `wss://` URL),
+ `LIVEKIT_KEY` / `LIVEKIT_SECRET` (LiveKit's API key) and
+ `LIVEKIT_FULL_ACCESS_HOMESERVERS` (your server name). The Helm chart
+ runs it with `calls.jwtService.enabled`.
+- **Warp** with `client_api.rtc.livekit_service_url` set to the JWT
+ service's URL. The JWT service must reach
+ `https:///.well-known/matrix/server` (a delegation to the
+ host that serves Warp) or `:8448`. Note that Warp serves
+ `/.well-known/matrix/server` only with federation on; with it off,
+ serve the delegation file from the server name's web server, as you
+ would for `/.well-known/matrix/client`.
+- **warp-web**: `WARP_WEB_CALL_ORIGINS="https://livekit-jwt.example.org wss://livekit.example.org"`
+ lets the browser reach both through its Content-Security-Policy.
+
+### Known gaps
+
+- Calls only use the legacy `org.matrix.msc3401.call.member` state events,
+ not the sticky `m.rtc.member` events of newer MatrixRTC drafts.
+- `org.matrix.msc4140.parent_delay_id` (chained delayed events) is
+ refused, and a delayed send's transaction ID is not deduplicated.
+- No ring for calls in rooms of more than two (only the banner), no
+ MSC4354 sticky events, and no MSC4195 homeserver-side LiveKit token
+ endpoint (`rtc/livekit/get_token`).
+- Calls from clients that only do legacy 1:1 VoIP (`m.call.*`) don't ring
+ in the web client.
diff --git a/docs/configuration.md b/docs/configuration.md
index 79d5734..fc5760f 100644
--- a/docs/configuration.md
+++ b/docs/configuration.md
@@ -62,10 +62,13 @@ List-valued variables are comma-separated. A single `-` clears a list:
| `client_api.rate_limit.join.burst` / `.interval` | — | `10` / `1s` | The two join endpoints. |
| `client_api.rate_limit.sync.burst` / `.interval` | — | `100` / `20ms` | `/sync` and the writes behind it: typing, receipts, to-device, presence. |
| `client_api.rate_limit.media.burst` / `.interval` | — | `100` / `20ms` | Media downloads and thumbnails: a client opening a room asks for every avatar at once. Uploads are in the default bucket. |
-| `client_api.turn.uris` | `WARP_TURN_URIS` | `[]` | `turn:`, `turns:` and `stun:` URIs for calls. Empty = no relay; `/voip/turnServer` answers `{}`. |
+| `client_api.turn.uris` | `WARP_TURN_URIS` | `[]` | `turn:`, `turns:` and `stun:` URIs for other clients' legacy 1:1 VoIP calls (the web client calls through LiveKit and doesn't use them). Empty = no relay; `/voip/turnServer` answers `{}`. |
| `client_api.turn.shared_secret` | `WARP_TURN_SHARED_SECRET` | — | coturn's `static-auth-secret`: each user gets expiring TURN REST API credentials. |
| `client_api.turn.user_lifetime` | `WARP_TURN_USER_LIFETIME` | `1h` | How long shared-secret credentials last. |
| `client_api.turn.username` / `.password` | `WARP_TURN_USERNAME` / `WARP_TURN_PASSWORD` | — | Static credentials, used only without a shared secret. |
+| `client_api.rtc.livekit_service_url` | `WARP_RTC_LIVEKIT_SERVICE_URL` | — | Base URL of the LiveKit JWT service (lk-jwt-service) for calls, advertised in `/.well-known/matrix/client` as `org.matrix.msc4143.rtc_foci`. Empty = no calls. See [calls.md](calls.md). |
+| `client_api.delayed_events.max_delay` | `WARP_DELAYED_EVENTS_MAX_DELAY` | `24h` | Longest delay a client may give a delayed event (MSC4140). `0` turns delayed events off and stops advertising `org.matrix.msc4140`; calls then can't remove a member whose client vanished. |
+| `client_api.delayed_events.max_per_user` | `WARP_DELAYED_EVENTS_MAX_PER_USER` | `100` | How many delayed events one user may have scheduled at once. |
| `sync.presence` | `WARP_SYNC_PRESENCE` | `false` | Off by default; see below. |
| `sync.typing_timeout` | `WARP_SYNC_TYPING_TIMEOUT` | `30s` | How long a typing notification lasts, and the cap on what a client may ask for. |
| `sync.long_poll_timeout` | `WARP_SYNC_LONG_POLL_TIMEOUT` | `30s` | The cap on a `/sync` `?timeout=`. |
diff --git a/docs/deployment.md b/docs/deployment.md
index e439a89..26a75ab 100644
--- a/docs/deployment.md
+++ b/docs/deployment.md
@@ -135,6 +135,21 @@ already public behind its own proxy, set `WARP_WEB_PUBLIC_HOMESERVER` to
its URL instead: the browser then calls it directly and warp-web serves
only the client. See [apps/web](../apps/web/README.md).
+## Calls
+
+Calls need a LiveKit SFU and the LiveKit JWT service next to Warp.
+For a local try, `deploy/compose/calls.yml` adds both:
+
+```sh
+cd deploy/compose
+WARP_REGISTRATION_ENABLED=true docker compose -f docker-compose.yml -f calls.yml --profile web up -d --build
+```
+
+The Helm chart can run the JWT service (`calls.jwtService.enabled`) and
+sets `client_api.rtc.livekit_service_url` from it; LiveKit itself runs
+from its own chart. See [calls.md](calls.md) for how the pieces find each
+other and what production needs.
+
## Reverse proxy
Behind a proxy that terminates TLS:
diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml
index 8311747..b734d32 100644
--- a/pnpm-lock.yaml
+++ b/pnpm-lock.yaml
@@ -71,6 +71,9 @@ importers:
jsqr:
specifier: 1.4.0
version: 1.4.0
+ livekit-client:
+ specifier: 2.22.3
+ version: 2.22.3(@types/dom-mediacapture-record@1.0.22)
matrix-js-sdk:
specifier: ^44.0.0
version: 44.0.0
@@ -131,6 +134,9 @@ packages:
resolution: {integrity: sha512-Vj1jF3cPfxg7OAfoI7QnVKLoILlm2JF9pnVHrX8qx7AHMiYWT+NDAA7jChlNgRS4WTLc/fD1lXLmPixluj+3Gg==}
engines: {node: '>=6.9.0'}
+ '@bufbuild/protobuf@1.10.1':
+ resolution: {integrity: sha512-wJ8ReQbHxsAfXhrf9ixl0aYbZorRuOWpBNzm8pL8ftmSxQx/wnJD5Eg861NwJU/czy2VXFIebCeZnZrI9rktIQ==}
+
'@esbuild/aix-ppc64@0.28.2':
resolution: {integrity: sha512-XExcO+dvLKvVtNTibSTBej1NCAbaGhWn9Ww1ZPx80qsahhPFe/8jgWP0IchNe0F3HwkU7n8ejhH8bjonqht8mQ==}
engines: {node: '>=18'}
@@ -303,6 +309,12 @@ packages:
'@jridgewell/trace-mapping@0.3.31':
resolution: {integrity: sha512-zzNR+SdQSDJzc8joaeP8QQoCQr8NuYx2dIIytl1QeBEZHJ9uW6hebsrYgbz8hJwUQao3TWCMtmfV8Nu1twOLAw==}
+ '@livekit/mutex@1.1.1':
+ resolution: {integrity: sha512-EsshAucklmpuUAfkABPxJNhzj9v2sG7JuzFDL4ML1oJQSV14sqrpTYnsaOudMAw9yOaW53NU3QQTlUQoRs4czw==}
+
+ '@livekit/protocol@1.50.4':
+ resolution: {integrity: sha512-L1uggNQAqyY21smQY8AllyOYbcv9Me9TaxwuLytL1R8ck9nbYPmQLNwEDi3pOFGAMa5F8I2nUi2Jc59W5awxlA==}
+
'@matrix-org/matrix-sdk-crypto-wasm@18.9.0':
resolution: {integrity: sha512-NtLzNDTEfNASBjNCTGZHuk7N0jrQhS6Zx2CM857rCHZYYh/rpbKGQNrJ7azktm4S1P5Jy3WU3nwJTGohGw26qQ==}
engines: {node: '>= 18'}
@@ -531,6 +543,9 @@ packages:
peerDependencies:
vite: ^5.2.0 || ^6 || ^7 || ^8
+ '@types/dom-mediacapture-record@1.0.22':
+ resolution: {integrity: sha512-mUMZLK3NvwRLcAAT9qmcK+9p7tpU2FHdDsntR3YI4+GY88XrgG4XiE7u1Q2LAN2/FZOz/tdMDC3GQCR4T8nFuw==}
+
'@types/estree@1.0.9':
resolution: {integrity: sha512-GhdPgy1el4/ImP05X05Uw4cw2/M93BCUmnEvWZNStlCzEKME4Fkk+YpoA5OiHNQmoS7Cafb8Xa3Pya8m1Qrzeg==}
@@ -676,6 +691,9 @@ packages:
resolution: {integrity: sha512-AC/7JofJvZGrrneWNaEnJeOLUx+JlGt7tNa0wZiRPT4MY1wmfKjt2+6O2p2uz2+skll8OZZmJMNqeke7kKbNgQ==}
hasBin: true
+ jose@6.2.12:
+ resolution: {integrity: sha512-9NiFmJEex0sy2Dk58j2UGBSHgUs2ypF9eZSu4L6vjOX3Dp96Sw1F3uL+H+D1sx02jZZdzUT0HgvCy59CuvXcWw==}
+
jsqr@1.4.0:
resolution: {integrity: sha512-dxLob7q65Xg2DvstYkRpkYtmKm2sPJ9oFhrhmudT1dZvNFFTlroai3AWSpLey/w5vMcLBXRgOJsbXpdN9HzU/A==}
@@ -749,10 +767,19 @@ packages:
resolution: {integrity: sha512-NXYBzinNrblfraPGyrbPoD19C1h9lfI/1mzgWYvXUTe414Gz/X1FD2XBZSZM7rRTrMA8JL3OtAaGifrIKhQ5yQ==}
engines: {node: '>= 12.0.0'}
+ livekit-client@2.22.3:
+ resolution: {integrity: sha512-jw9zBKXY5Gtr5MZ7vEON3QhMNccuDvYHck1PFSyG1aaateQPqgKZFBMgZkFZaXHIf9RV4MDW5xpTK2b/+qbwOg==}
+ peerDependencies:
+ '@types/dom-mediacapture-record': ^1
+
loglevel@1.9.2:
resolution: {integrity: sha512-HgMmCqIJSAKqo68l0rS2AanEWfkxaZ5wNiEFb5ggm08lDs9Xl2KxBlX3PTcaD2chBM1gXAYf491/M2Rv8Jwayg==}
engines: {node: '>= 0.6.0'}
+ machina@7.0.1:
+ resolution: {integrity: sha512-0Xkgs8NeBvO2hW6bjJT8ewLWrMfh1haMPstaghpASWjXkLjRhtDiaJG1Kb7mslaOHxZYxpH7PIjpmirRVYbMMA==}
+ engines: {node: '>=22.22'}
+
magic-string@0.30.21:
resolution: {integrity: sha512-vd2F4YUyEXKGcLHoq+TEyCjxueSeHnFxyyjNp80yg0XV4vUhnDer/lvvlqM/arB5bXQN5K2/3oinyCRyx8T2CQ==}
@@ -818,10 +845,20 @@ packages:
engines: {node: '>=18.0.0', npm: '>=8.0.0'}
hasBin: true
+ rxjs@7.8.2:
+ resolution: {integrity: sha512-dhKf903U/PQZY6boNNtAGdWbG85WAbjT/1xYoZIC7FAY0yWapOBQVsVrDl58W86//e1VpMNBtRV4MaXfdMySFA==}
+
+ sdp-transform@2.15.0:
+ resolution: {integrity: sha512-KrOH82c/W+GYQ0LHqtr3caRpM3ITglq3ljGUIb8LTki7ByacJZ9z+piSGiwZDsRyhQbYBOBJgr2k6X4BZXi3Kw==}
+ hasBin: true
+
sdp-transform@3.0.0:
resolution: {integrity: sha512-gfYVRGxjHkGF2NPeUWHw5u6T/KGFtS5/drPms73gaSuMaVHKCY3lpLnGDfswVQO0kddeePoti09AwhYP4zA8dQ==}
hasBin: true
+ sdp@3.2.2:
+ resolution: {integrity: sha512-xZocWwfyp4hkbN4hLWxMjmv2Q8aNa9MhmOZ7L9aCZPT+dZsgRr6wZRrSYE3HTdyk/2pZKPSgqI7ns7Een1xMSA==}
+
source-map-js@1.2.2:
resolution: {integrity: sha512-KGj/8Y43x35aZVDtt+J4mK1hoLGHULMYfSkODJNQjNDC3oW1PqPoxMwo0pLUsWM/UEGzON/NxeHywEfNXNP3Vw==}
engines: {node: '>=0.10.0'}
@@ -845,6 +882,12 @@ packages:
resolution: {integrity: sha512-wXR/dYpcqKmfWpEdZjiKJOwCNFndD0DMnrW/cYjVGttEkBfVgcLFHoNrlj47mjOVic9yyNu65alsgF4NQyTa2g==}
engines: {node: '>=12.0.0'}
+ tslib@2.8.1:
+ resolution: {integrity: sha512-oJFu94HQb+KVduSUQL7wnpmqnfmLsOA/nAh6b6EH0wCEoK0/mPeXU6c3wKDV83MkOuHPRHtSXKKU99IBazS/2w==}
+
+ typed-emitter@2.1.0:
+ resolution: {integrity: sha512-g/KzbYKbH5C2vPkaXGu8DJlHrGKHLsM25Zg9WuC9pMGfuvT+X25tZQWo5fK1BjBm8+UrVE9LDCvaY0CQk+fXDA==}
+
typescript@5.9.3:
resolution: {integrity: sha512-jl1vZzPDinLr9eUt3J/t7V6FgNEw9QjvBPdysz9KfQDD41fQrC2Y4vKQdiaUpFT4bXlb1RHhLpp8wtm6M5TgSw==}
engines: {node: '>=14.17'}
@@ -918,6 +961,10 @@ packages:
typescript:
optional: true
+ webrtc-adapter@9.0.6:
+ resolution: {integrity: sha512-CHbl2ZQbxx164IgWRgzJno4hWtM4tFbRam1QfI3Yxhs3w/DvqluVxVWeXs3oL5/fbGkSNLKo0Ty5MgUWceNhog==}
+ engines: {node: '>=6.0.0', npm: '>=3.10.0'}
+
snapshots:
'@babel/helper-string-parser@7.29.7': {}
@@ -935,6 +982,8 @@ snapshots:
'@babel/helper-string-parser': 7.29.7
'@babel/helper-validator-identifier': 7.29.7
+ '@bufbuild/protobuf@1.10.1': {}
+
'@esbuild/aix-ppc64@0.28.2':
optional: true
@@ -1032,6 +1081,12 @@ snapshots:
'@jridgewell/resolve-uri': 3.1.2
'@jridgewell/sourcemap-codec': 1.6.0
+ '@livekit/mutex@1.1.1': {}
+
+ '@livekit/protocol@1.50.4':
+ dependencies:
+ '@bufbuild/protobuf': 1.10.1
+
'@matrix-org/matrix-sdk-crypto-wasm@18.9.0': {}
'@napi-rs/lzma-linux-x64-gnu@1.5.1':
@@ -1182,6 +1237,8 @@ snapshots:
tailwindcss: 4.3.3
vite: 7.3.7(@types/node@22.20.5)(jiti@2.7.0)(lightningcss@1.32.0)
+ '@types/dom-mediacapture-record@1.0.22': {}
+
'@types/estree@1.0.9': {}
'@types/events@3.0.3': {}
@@ -1367,6 +1424,8 @@ snapshots:
jiti@2.7.0: {}
+ jose@6.2.12: {}
+
jsqr@1.4.0: {}
lightningcss-android-arm64@1.32.0:
@@ -1418,8 +1477,24 @@ snapshots:
lightningcss-win32-arm64-msvc: 1.32.0
lightningcss-win32-x64-msvc: 1.32.0
+ livekit-client@2.22.3(@types/dom-mediacapture-record@1.0.22):
+ dependencies:
+ '@livekit/mutex': 1.1.1
+ '@livekit/protocol': 1.50.4
+ '@types/dom-mediacapture-record': 1.0.22
+ events: 3.3.0
+ jose: 6.2.12
+ loglevel: 1.9.2
+ machina: 7.0.1
+ sdp-transform: 2.15.0
+ tslib: 2.8.1
+ typed-emitter: 2.1.0
+ webrtc-adapter: 9.0.6
+
loglevel@1.9.2: {}
+ machina@7.0.1: {}
+
magic-string@0.30.21:
dependencies:
'@jridgewell/sourcemap-codec': 1.6.0
@@ -1512,8 +1587,17 @@ snapshots:
'@rollup/rollup-win32-x64-msvc': 4.64.9
fsevents: 2.3.3
+ rxjs@7.8.2:
+ dependencies:
+ tslib: 2.8.1
+ optional: true
+
+ sdp-transform@2.15.0: {}
+
sdp-transform@3.0.0: {}
+ sdp@3.2.2: {}
+
source-map-js@1.2.2: {}
speakingurl@14.0.1: {}
@@ -1531,6 +1615,12 @@ snapshots:
fdir: 6.5.0(picomatch@4.0.7)
picomatch: 4.0.7
+ tslib@2.8.1: {}
+
+ typed-emitter@2.1.0:
+ optionalDependencies:
+ rxjs: 7.8.2
+
typescript@5.9.3: {}
undici-types@6.21.0: {}
@@ -1573,3 +1663,7 @@ snapshots:
'@vue/shared': 3.5.44
optionalDependencies:
typescript: 5.9.3
+
+ webrtc-adapter@9.0.6:
+ dependencies:
+ sdp: 3.2.2
diff --git a/server/internal/clientapi/clientapi.go b/server/internal/clientapi/clientapi.go
index e802ac9..7708c13 100644
--- a/server/internal/clientapi/clientapi.go
+++ b/server/internal/clientapi/clientapi.go
@@ -19,6 +19,7 @@ import (
"github.com/dariuszjaniak/warp/server/internal/appservice"
"github.com/dariuszjaniak/warp/server/internal/config"
+ "github.com/dariuszjaniak/warp/server/internal/delayedevents"
"github.com/dariuszjaniak/warp/server/internal/federation"
"github.com/dariuszjaniak/warp/server/internal/httputil"
"github.com/dariuszjaniak/warp/server/internal/keyserver"
@@ -27,6 +28,7 @@ import (
"github.com/dariuszjaniak/warp/server/internal/ratelimit"
"github.com/dariuszjaniak/warp/server/internal/roomserver"
"github.com/dariuszjaniak/warp/server/internal/settings"
+ "github.com/dariuszjaniak/warp/server/internal/storage"
"github.com/dariuszjaniak/warp/server/internal/syncapi"
"github.com/dariuszjaniak/warp/server/internal/userapi"
)
@@ -48,6 +50,9 @@ type ClientAPI struct {
regTokens *userapi.RegistrationTokens
appservices *appservice.API
turn config.TURN
+ delayed *delayedevents.Service
+ rtcTransports []RTCTransport
+ openID storage.OpenIDTokenStore
log *slog.Logger
now func() time.Time
@@ -145,6 +150,20 @@ type Options struct {
// TURN is the relay handed to clients for calls.
TURN config.TURN
+ // RTC is the group call backend, listed at the MSC4143 transports
+ // endpoint.
+ RTC config.RTC
+
+ // DelayedEvents schedules MSC4140 delayed events. Nil, or a service
+ // with delayed events switched off, leaves the endpoints unregistered
+ // and refuses the delay parameter.
+ DelayedEvents *delayedevents.Service
+
+ // OpenIDTokens keeps the OpenID tokens a user hands third parties,
+ // such as the LiveKit JWT service of a group call. Nil leaves the
+ // OpenID endpoints unregistered.
+ OpenIDTokens storage.OpenIDTokenStore
+
// TrustXForwardedFor takes the client address from the proxy header.
TrustXForwardedFor bool
@@ -205,6 +224,9 @@ func New(opts Options) (*ClientAPI, error) {
regTokens: opts.RegistrationTokens,
appservices: opts.AppServices,
turn: opts.TURN,
+ delayed: opts.DelayedEvents,
+ rtcTransports: RTCTransports(opts.RTC.LiveKitServiceURL),
+ openID: opts.OpenIDTokens,
log: opts.Logger,
now: opts.Now,
uia: newUIAStore(uiaSessionTTL, opts.Now),
@@ -274,6 +296,9 @@ func (c *ClientAPI) RegisterRoutes(r *httputil.Router) {
// Calls: the relay a client sets its peer connection up with.
c.authed(r, "GET "+v3+"/voip/turnServer", c.limitDefault, c.handleTurnServer)
+ c.registerOpenIDRoutes(r)
+ c.registerDelayedEventRoutes(r)
+ c.registerRTCRoutes(r)
// Devices.
c.authed(r, "GET "+v3+"/devices", c.limitDefault, c.handleGetDevices)
diff --git a/server/internal/clientapi/clientapi_test.go b/server/internal/clientapi/clientapi_test.go
index 92b7fe8..d9e0828 100644
--- a/server/internal/clientapi/clientapi_test.go
+++ b/server/internal/clientapi/clientapi_test.go
@@ -2,16 +2,20 @@ package clientapi_test
import (
"bytes"
+ "context"
"encoding/json"
"io"
"log/slog"
"net/http"
"net/http/httptest"
"path/filepath"
+ "sync"
"testing"
+ "time"
"github.com/dariuszjaniak/warp/server/internal/clientapi"
"github.com/dariuszjaniak/warp/server/internal/config"
+ "github.com/dariuszjaniak/warp/server/internal/delayedevents"
"github.com/dariuszjaniak/warp/server/internal/federation"
"github.com/dariuszjaniak/warp/server/internal/httputil"
"github.com/dariuszjaniak/warp/server/internal/matrix/signing"
@@ -39,6 +43,44 @@ type testServer struct {
settings *settings.Service
fedGate *federation.Gate
db storage.Database
+
+ // delayed is the delayed event service, on a clock of its own that
+ // tests move with advance.
+ delayed *delayedevents.Service
+ clock *testClock
+}
+
+// testClock is a clock a test moves by hand.
+type testClock struct {
+ mu sync.Mutex
+ now time.Time
+}
+
+func (c *testClock) Now() time.Time {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ return c.now
+}
+
+func (c *testClock) advance(d time.Duration) {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ c.now = c.now.Add(d)
+}
+
+// teeNotifier passes the roomserver's emit stage on to the syncapi and,
+// once it is built, to the delayed event service, which needs the
+// roomserver to exist first.
+type teeNotifier struct {
+ first roomserver.Notifier
+ then roomserver.Notifier
+}
+
+func (n *teeNotifier) EventPersisted(ctx context.Context, note roomserver.Notification) {
+ n.first.EventPersisted(ctx, note)
+ if n.then != nil {
+ n.then.EventPersisted(ctx, note)
+ }
}
// newTestServer builds a server. tweak adjusts the options before the
@@ -91,13 +133,14 @@ func newTestServer(t *testing.T, tweak func(*clientapi.Options)) *testServer {
// stage and the syncapi's wake-up are the same object, exactly as in
// a real server.
notifier := syncapi.NewNotifier()
+ emit := &teeNotifier{first: notifier}
rooms, err := roomserver.New(roomserver.Options{
ServerName: testServerName,
Store: db.Rooms(),
Profiles: users,
SigningKey: key,
FederationEnabled: settingsService.FederationEnabled,
- Notifier: notifier,
+ Notifier: emit,
Logger: slog.New(slog.DiscardHandler),
})
if err != nil {
@@ -121,14 +164,34 @@ func newTestServer(t *testing.T, tweak func(*clientapi.Options)) *testServer {
}
t.Cleanup(sync.Close)
+ // Delayed events are on, with the scheduler left stopped: a test
+ // moves the clock and calls SendDue itself.
+ clock := &testClock{now: time.Now()}
+ delayed, err := delayedevents.New(delayedevents.Options{
+ Store: db.DelayedEvents(),
+ Rooms: rooms,
+ MaxDelay: time.Hour,
+ MaxPerUser: 5,
+ DescribeError: clientapi.DescribeRoomError,
+ Logger: slog.New(slog.DiscardHandler),
+ Now: clock.Now,
+ })
+ if err != nil {
+ t.Fatalf("delayedevents.New: %v", err)
+ }
+ t.Cleanup(delayed.Close)
+ emit.then = delayed
+
opts := clientapi.Options{
- ServerName: testServerName,
- Users: users,
- Rooms: rooms,
- Sync: sync,
- Settings: settingsService,
- Federation: gate,
- Logger: slog.New(slog.DiscardHandler),
+ ServerName: testServerName,
+ Users: users,
+ Rooms: rooms,
+ Sync: sync,
+ Settings: settingsService,
+ Federation: gate,
+ DelayedEvents: delayed,
+ OpenIDTokens: db.OpenIDTokens(),
+ Logger: slog.New(slog.DiscardHandler),
}
if tweak != nil {
tweak(&opts)
@@ -150,6 +213,8 @@ func newTestServer(t *testing.T, tweak func(*clientapi.Options)) *testServer {
settings: settingsService,
fedGate: gate,
db: db,
+ delayed: delayed,
+ clock: clock,
}
}
diff --git a/server/internal/clientapi/delayed.go b/server/internal/clientapi/delayed.go
new file mode 100644
index 0000000..47dd0c5
--- /dev/null
+++ b/server/internal/clientapi/delayed.go
@@ -0,0 +1,259 @@
+package clientapi
+
+import (
+ "encoding/json"
+ "errors"
+ "net/http"
+ "slices"
+ "strconv"
+ "time"
+
+ "github.com/dariuszjaniak/warp/server/internal/delayedevents"
+ "github.com/dariuszjaniak/warp/server/internal/httputil"
+ "github.com/dariuszjaniak/warp/server/internal/roomserver"
+ "github.com/dariuszjaniak/warp/server/internal/storage"
+ "github.com/dariuszjaniak/warp/server/internal/userapi"
+)
+
+// MSC4140 is still unstable, so its endpoints and parameters carry its
+// prefix.
+const (
+ delayedEventsPrefix = "/_matrix/client/unstable/org.matrix.msc4140"
+ delayParam = "org.matrix.msc4140.delay"
+ parentDelayParam = "org.matrix.msc4140.parent_delay_id"
+)
+
+// registerDelayedEventRoutes adds the MSC4140 endpoints, if delayed events
+// are switched on.
+//
+// The action endpoints are public: the delay ID is the capability, which is
+// what lets a client hand its delayed leave to something else to keep alive.
+// Both spellings are served — the action in the path, which is the current
+// draft, and in the body, which clients fall back to.
+func (c *ClientAPI) registerDelayedEventRoutes(r *httputil.Router) {
+ if !c.delayed.Enabled() || c.rooms == nil {
+ return
+ }
+ c.authed(r, "GET "+delayedEventsPrefix+"/delayed_events", c.limitDefault, c.handleGetDelayedEvents)
+ c.authed(r, "GET "+delayedEventsPrefix+"/delayed_events/{delayID}", c.limitDefault, c.handleGetDelayedEvent)
+ c.public(r, "POST "+delayedEventsPrefix+"/delayed_events/{delayID}", c.limitDefault, c.handleUpdateDelayedEvent)
+ c.public(r, "POST "+delayedEventsPrefix+"/delayed_events/{delayID}/{action}", c.limitDefault, c.handleUpdateDelayedEvent)
+}
+
+// delayRequested reads the delay parameter of a send or state PUT. ok is
+// false when there is none, and the event is sent at once as usual.
+func (c *ClientAPI) delayRequested(r *http.Request) (time.Duration, bool, error) {
+ q := r.URL.Query()
+ if q.Has(parentDelayParam) {
+ return 0, false, httputil.InvalidParam("%s is not supported", parentDelayParam)
+ }
+ if !q.Has(delayParam) {
+ return 0, false, nil
+ }
+ if !c.delayed.Enabled() {
+ // Refused rather than ignored: sending a delayed "left the call"
+ // at once would be exactly the wrong thing.
+ return 0, false, httputil.InvalidParam("Delayed events are not enabled on this server")
+ }
+ ms, err := strconv.ParseInt(q.Get(delayParam), 10, 64)
+ if err != nil || ms < 0 {
+ return 0, false, httputil.InvalidParam("%s must be a non-negative number of milliseconds", delayParam)
+ }
+ return time.Duration(ms) * time.Millisecond, true, nil
+}
+
+// delayIDResponse answers a delayed send.
+type delayIDResponse struct {
+ DelayID string `json:"delay_id"`
+}
+
+// scheduleDelayed schedules what a send or state PUT asked to delay.
+func (c *ClientAPI) scheduleDelayed(w http.ResponseWriter, r *http.Request, req delayedevents.ScheduleRequest) error {
+ delayID, err := c.delayed.Schedule(r.Context(), req)
+ if err != nil {
+ return delayedError(err, c.delayed.MaxDelay())
+ }
+ httputil.WriteJSON(w, http.StatusOK, delayIDResponse{DelayID: delayID})
+ return nil
+}
+
+// delayedError maps the service's errors onto Matrix errors. An overlong
+// delay is reported the way Synapse does: M_UNKNOWN for a client that knows
+// nothing of MSC4140, with the MSC's own code and the maximum alongside.
+func delayedError(err error, maxDelay time.Duration) error {
+ switch {
+ case errors.Is(err, delayedevents.ErrMaxDelayExceeded):
+ merr := httputil.NewError(http.StatusBadRequest, httputil.ErrCodeUnknown, "%s", err.Error())
+ merr.DelayedErrCode = "M_MAX_DELAY_EXCEEDED"
+ merr.MaxDelayMS = maxDelay.Milliseconds()
+ return merr
+ case errors.Is(err, delayedevents.ErrTooMany):
+ return httputil.NewError(http.StatusBadRequest, "M_LIMIT_EXCEEDED", "%s", err.Error())
+ case errors.Is(err, delayedevents.ErrDisabled):
+ return httputil.Unrecognized()
+ case errors.Is(err, storage.ErrNoSuchDelayedEvent):
+ return httputil.NotFound("No such scheduled delayed event")
+ default:
+ return roomError(err)
+ }
+}
+
+// delayedEventJSON is one delayed event as MSC4140 lists it.
+type delayedEventJSON struct {
+ DelayID string `json:"delay_id"`
+ RoomID string `json:"room_id"`
+ Type string `json:"type"`
+ StateKey *string `json:"state_key,omitempty"`
+ Content json.RawMessage `json:"content"`
+ Delay int64 `json:"delay"`
+ RunningSince int64 `json:"running_since"`
+}
+
+type finalisedDelayedEventJSON struct {
+ DelayedEvent delayedEventJSON `json:"delayed_event"`
+ Outcome string `json:"outcome"`
+ Reason string `json:"reason"`
+ Error json.RawMessage `json:"error,omitempty"`
+ EventID string `json:"event_id,omitempty"`
+ OriginServerTS int64 `json:"origin_server_ts,omitempty"`
+}
+
+type delayedEventsResponse struct {
+ Scheduled []delayedEventJSON `json:"scheduled"`
+ Finalised []finalisedDelayedEventJSON `json:"finalised"`
+
+ // DelayedEvents is the scheduled list under the name an earlier draft
+ // of MSC4140 used.
+ DelayedEvents []delayedEventJSON `json:"delayed_events"`
+}
+
+func toDelayedEventJSON(e storage.DelayedEvent) delayedEventJSON {
+ return delayedEventJSON{
+ DelayID: e.DelayID,
+ RoomID: e.RoomID,
+ Type: e.Type,
+ StateKey: e.StateKey,
+ Content: e.Content,
+ Delay: e.DelayMS,
+ RunningSince: e.RunningSinceMS,
+ }
+}
+
+func toFinalisedJSON(e storage.DelayedEvent) finalisedDelayedEventJSON {
+ out := finalisedDelayedEventJSON{
+ DelayedEvent: toDelayedEventJSON(e),
+ Outcome: e.Outcome,
+ Reason: e.Reason,
+ Error: e.Error,
+ EventID: e.EventID,
+ }
+ if e.EventID != "" {
+ out.OriginServerTS = e.FinalisedAtMS
+ }
+ return out
+}
+
+// handleGetDelayedEvents implements
+// GET /_matrix/client/unstable/org.matrix.msc4140/delayed_events.
+//
+// status narrows the answer to scheduled or finalised events, and delay_id,
+// which may repeat, to particular ones. Nobody has enough delayed events
+// for paging to matter, so there is never a next_batch.
+func (c *ClientAPI) handleGetDelayedEvents(w http.ResponseWriter, r *http.Request, auth userapi.Authenticated) error {
+ q := r.URL.Query()
+ status := q.Get("status")
+ if status != "" && status != "scheduled" && status != "finalised" {
+ return httputil.InvalidParam("status must be scheduled or finalised")
+ }
+ ids := q["delay_id"]
+
+ events, err := c.delayed.List(r.Context(), auth.UserID)
+ if err != nil {
+ return err
+ }
+ resp := delayedEventsResponse{
+ Scheduled: []delayedEventJSON{},
+ Finalised: []finalisedDelayedEventJSON{},
+ }
+ for _, e := range events {
+ if len(ids) > 0 && !slices.Contains(ids, e.DelayID) {
+ continue
+ }
+ if e.FinalisedAtMS == 0 {
+ if status != "finalised" {
+ resp.Scheduled = append(resp.Scheduled, toDelayedEventJSON(e))
+ }
+ } else if status != "scheduled" {
+ resp.Finalised = append(resp.Finalised, toFinalisedJSON(e))
+ }
+ }
+ resp.DelayedEvents = resp.Scheduled
+ httputil.WriteJSON(w, http.StatusOK, resp)
+ return nil
+}
+
+// handleGetDelayedEvent implements
+// GET /_matrix/client/unstable/org.matrix.msc4140/delayed_events/{delayID}.
+func (c *ClientAPI) handleGetDelayedEvent(w http.ResponseWriter, r *http.Request, auth userapi.Authenticated) error {
+ e, err := c.delayed.Get(r.Context(), auth.UserID, r.PathValue("delayID"))
+ if err != nil {
+ return delayedError(err, 0)
+ }
+ if e.FinalisedAtMS == 0 {
+ httputil.WriteJSON(w, http.StatusOK, toDelayedEventJSON(e))
+ } else {
+ httputil.WriteJSON(w, http.StatusOK, toFinalisedJSON(e))
+ }
+ return nil
+}
+
+type updateDelayedEventRequest struct {
+ Action string `json:"action"`
+}
+
+// handleUpdateDelayedEvent implements
+// POST /_matrix/client/unstable/org.matrix.msc4140/delayed_events/{delayID}[/{action}].
+func (c *ClientAPI) handleUpdateDelayedEvent(w http.ResponseWriter, r *http.Request) error {
+ delayID := r.PathValue("delayID")
+ action := r.PathValue("action")
+ if action == "" {
+ var req updateDelayedEventRequest
+ if err := httputil.DecodeJSON(w, r, &req); err != nil {
+ return err
+ }
+ action = req.Action
+ }
+
+ var err error
+ switch action {
+ case "restart":
+ err = c.delayed.Restart(r.Context(), delayID)
+ case "cancel":
+ err = c.delayed.Cancel(r.Context(), delayID)
+ case "send":
+ err = c.delayed.Send(r.Context(), delayID)
+ case "":
+ return httputil.MissingParam("An action is required")
+ default:
+ return httputil.InvalidParam("action must be restart, cancel or send")
+ }
+ if err != nil {
+ return delayedError(err, 0)
+ }
+ httputil.WriteJSON(w, http.StatusOK, struct{}{})
+ return nil
+}
+
+// DescribeRoomError is the Matrix error a failed delayed send is recorded
+// with: the error the client would have got had it sent the event itself.
+func DescribeRoomError(err error) json.RawMessage {
+ var matrixErr *httputil.MatrixError
+ if !errors.As(roomError(err), &matrixErr) {
+ matrixErr = httputil.NewError(http.StatusInternalServerError, httputil.ErrCodeUnknown, "%s", err.Error())
+ }
+ raw, _ := json.Marshal(matrixErr)
+ return raw
+}
+
+// Compile-time check that the roomserver can be the service's sender.
+var _ delayedevents.Sender = (*roomserver.API)(nil)
diff --git a/server/internal/clientapi/delayed_test.go b/server/internal/clientapi/delayed_test.go
new file mode 100644
index 0000000..b23da27
--- /dev/null
+++ b/server/internal/clientapi/delayed_test.go
@@ -0,0 +1,300 @@
+package clientapi_test
+
+import (
+ "net/http"
+ "net/url"
+ "strconv"
+ "testing"
+ "time"
+
+ "github.com/dariuszjaniak/warp/server/internal/clientapi"
+ "github.com/dariuszjaniak/warp/server/internal/httputil"
+)
+
+const delayedEventsPath = "/_matrix/client/unstable/org.matrix.msc4140/delayed_events"
+
+// callMemberPath is where a member of a room's group call puts its state.
+func callMemberPath(roomID, stateKey string) string {
+ return "/_matrix/client/v3/rooms/" + url.PathEscape(roomID) +
+ "/state/org.matrix.msc3401.call.member/" + url.PathEscape(stateKey)
+}
+
+// scheduleDelayed PUTs a delayed state event and returns its delay ID.
+func (ts *testServer) scheduleDelayed(path string, delayMS int, acc account, content map[string]any) string {
+ ts.t.Helper()
+ rec := ts.put(path+"?org.matrix.msc4140.delay="+strconv.Itoa(delayMS), acc.AccessToken, content)
+ expectStatus(ts.t, rec, http.StatusOK)
+ delayID, _ := jsonBody(ts.t, rec)["delay_id"].(string)
+ if delayID == "" {
+ ts.t.Fatalf("a delayed send returned no delay_id: %s", rec.Body.String())
+ }
+ return delayID
+}
+
+// tick moves the clock and sends whatever has come due.
+func (ts *testServer) tick(d time.Duration) {
+ ts.t.Helper()
+ ts.clock.advance(d)
+ ts.delayed.SendDue(ts.t.Context())
+}
+
+type delayedList struct {
+ Scheduled []struct {
+ DelayID string `json:"delay_id"`
+ RoomID string `json:"room_id"`
+ Type string `json:"type"`
+ StateKey *string `json:"state_key"`
+ Content map[string]any `json:"content"`
+ Delay int64 `json:"delay"`
+ RunningSince int64 `json:"running_since"`
+ } `json:"scheduled"`
+ Finalised []struct {
+ DelayedEvent struct {
+ DelayID string `json:"delay_id"`
+ } `json:"delayed_event"`
+ Outcome string `json:"outcome"`
+ Reason string `json:"reason"`
+ EventID string `json:"event_id"`
+ Error map[string]any `json:"error"`
+ } `json:"finalised"`
+}
+
+func (ts *testServer) delayedList(acc account) delayedList {
+ ts.t.Helper()
+ rec := ts.get(delayedEventsPath, acc.AccessToken)
+ expectStatus(ts.t, rec, http.StatusOK)
+ return decode[delayedList](ts.t, rec)
+}
+
+// TestDelayedStateEventIsSentWhenDue is the MatrixRTC case: a member's
+// "left the call" is scheduled when it joins, and arrives on its own once
+// the delay runs out.
+func TestDelayedStateEventIsSentWhenDue(t *testing.T) {
+ ts := newTestServer(t, nil)
+ alice := ts.register("alice", "correct-horse-battery-staple")
+ roomID := ts.createRoom(alice, map[string]any{"preset": "public_chat"})
+ stateKey := "_" + alice.UserID + "_" + alice.DeviceID + "_m.call"
+ path := callMemberPath(roomID, stateKey)
+
+ delayID := ts.scheduleDelayed(path, 5000, alice, map[string]any{})
+
+ // Nothing is in the room yet.
+ expectError(t, ts.get(path, alice.AccessToken), http.StatusNotFound, httputil.ErrCodeNotFound)
+
+ list := ts.delayedList(alice)
+ if len(list.Scheduled) != 1 || list.Scheduled[0].DelayID != delayID {
+ t.Fatalf("scheduled = %+v", list.Scheduled)
+ }
+ got := list.Scheduled[0]
+ if got.RoomID != roomID || got.Type != "org.matrix.msc3401.call.member" ||
+ got.StateKey == nil || *got.StateKey != stateKey || got.Delay != 5000 {
+ t.Errorf("the scheduled event is %+v", got)
+ }
+
+ ts.tick(4 * time.Second)
+ expectError(t, ts.get(path, alice.AccessToken), http.StatusNotFound, httputil.ErrCodeNotFound)
+
+ ts.tick(2 * time.Second)
+ expectStatus(t, ts.get(path, alice.AccessToken), http.StatusOK)
+
+ list = ts.delayedList(alice)
+ if len(list.Scheduled) != 0 {
+ t.Errorf("still scheduled after sending: %+v", list.Scheduled)
+ }
+ if len(list.Finalised) != 1 || list.Finalised[0].Outcome != "send" ||
+ list.Finalised[0].Reason != "delay" || list.Finalised[0].EventID == "" {
+ t.Errorf("finalised = %+v", list.Finalised)
+ }
+
+ // Looked up on its own, it is the same.
+ rec := ts.get(delayedEventsPath+"/"+delayID, alice.AccessToken)
+ expectStatus(t, rec, http.StatusOK)
+ if jsonBody(t, rec)["outcome"] != "send" {
+ t.Errorf("GET by ID = %s", rec.Body.String())
+ }
+}
+
+// TestDelayedEventRestart: a client still in the call keeps pushing its
+// leave back, and it never arrives.
+func TestDelayedEventRestart(t *testing.T) {
+ ts := newTestServer(t, nil)
+ alice := ts.register("alice", "correct-horse-battery-staple")
+ roomID := ts.createRoom(alice, map[string]any{"preset": "public_chat"})
+ path := callMemberPath(roomID, "_"+alice.UserID+"_D_m.call")
+
+ delayID := ts.scheduleDelayed(path, 5000, alice, map[string]any{})
+
+ for range 3 {
+ ts.tick(4 * time.Second)
+ // The action endpoints need no token: the delay ID is the
+ // capability.
+ expectStatus(t, ts.post(delayedEventsPath+"/"+delayID+"/restart", "", nil), http.StatusOK)
+ }
+ ts.tick(4 * time.Second)
+ expectError(t, ts.get(path, alice.AccessToken), http.StatusNotFound, httputil.ErrCodeNotFound)
+
+ ts.tick(2 * time.Second)
+ expectStatus(t, ts.get(path, alice.AccessToken), http.StatusOK)
+
+ // Once it has been sent it cannot be restarted.
+ rec := ts.post(delayedEventsPath+"/"+delayID+"/restart", "", nil)
+ expectError(t, rec, http.StatusNotFound, httputil.ErrCodeNotFound)
+}
+
+// TestDelayedEventActions covers cancel and send, in both the path and the
+// body spelling.
+func TestDelayedEventActions(t *testing.T) {
+ ts := newTestServer(t, nil)
+ alice := ts.register("alice", "correct-horse-battery-staple")
+ roomID := ts.createRoom(alice, map[string]any{"preset": "public_chat"})
+ escaped := url.PathEscape(roomID)
+
+ // Cancelled with the action in the body: never sent.
+ topic := "/_matrix/client/v3/rooms/" + escaped + "/state/m.room.topic"
+ cancelled := ts.scheduleDelayed(topic, 1000, alice, map[string]any{"topic": "never"})
+ expectStatus(t, ts.post(delayedEventsPath+"/"+cancelled, alice.AccessToken,
+ map[string]any{"action": "cancel"}), http.StatusOK)
+ ts.tick(2 * time.Second)
+ expectError(t, ts.get(topic, alice.AccessToken), http.StatusNotFound, httputil.ErrCodeNotFound)
+
+ // Sent at once with the action in the path: a timeline event, which
+ // is how a client sends a message it held back.
+ send := "/_matrix/client/v3/rooms/" + escaped + "/send/m.room.message/txn1"
+ sent := ts.scheduleDelayed(send, 60000, alice, map[string]any{"msgtype": "m.text", "body": "later"})
+ expectStatus(t, ts.post(delayedEventsPath+"/"+sent+"/send", "", nil), http.StatusOK)
+
+ list := ts.delayedList(alice)
+ if len(list.Scheduled) != 0 || len(list.Finalised) != 2 {
+ t.Fatalf("list = %+v", list)
+ }
+ outcomes := map[string]string{}
+ for _, f := range list.Finalised {
+ outcomes[f.DelayedEvent.DelayID] = f.Outcome + "/" + f.Reason
+ }
+ if outcomes[cancelled] != "cancel/action" || outcomes[sent] != "send/action" {
+ t.Errorf("outcomes = %v", outcomes)
+ }
+
+ // status narrows the list.
+ rec := ts.get(delayedEventsPath+"?status=scheduled", alice.AccessToken)
+ expectStatus(t, rec, http.StatusOK)
+ if l := decode[delayedList](t, rec); len(l.Finalised) != 0 {
+ t.Errorf("status=scheduled listed finalised events: %s", rec.Body.String())
+ }
+
+ expectError(t, ts.post(delayedEventsPath+"/nope/cancel", "", nil), http.StatusNotFound, httputil.ErrCodeNotFound)
+ expectError(t, ts.post(delayedEventsPath+"/"+sent+"/explode", "", nil),
+ http.StatusBadRequest, httputil.ErrCodeInvalidParam)
+}
+
+// TestDelayedEventFailureIsRecorded: an event that can no longer be sent
+// is finalised with the error the client would have got.
+func TestDelayedEventFailureIsRecorded(t *testing.T) {
+ ts := newTestServer(t, nil)
+ alice := ts.register("alice", "correct-horse-battery-staple")
+ roomID := ts.createRoom(alice, map[string]any{"preset": "public_chat"})
+ path := callMemberPath(roomID, "_"+alice.UserID+"_D_m.call")
+
+ ts.scheduleDelayed(path, 1000, alice, map[string]any{})
+ expectStatus(t, ts.post("/_matrix/client/v3/rooms/"+url.PathEscape(roomID)+"/leave",
+ alice.AccessToken, map[string]any{}), http.StatusOK)
+ ts.tick(2 * time.Second)
+
+ list := ts.delayedList(alice)
+ if len(list.Finalised) != 1 || list.Finalised[0].Reason != "error" ||
+ list.Finalised[0].Error["errcode"] != httputil.ErrCodeForbidden {
+ t.Errorf("finalised = %+v", list.Finalised)
+ }
+}
+
+// TestDelayedStateIsCancelledByOthers: someone else setting the same state
+// cancels a delayed event that would have overwritten it, and the owner's
+// own updates do not. It also proves an ordinary member may join a call.
+func TestDelayedStateIsCancelledByOthers(t *testing.T) {
+ ts := newTestServer(t, nil)
+ alice := ts.register("alice", "correct-horse-battery-staple")
+ bob := ts.register("bob", "correct-horse-battery-staple")
+ roomID := ts.createRoom(alice, map[string]any{"preset": "public_chat"})
+ expectStatus(t, ts.post("/_matrix/client/v3/rooms/"+url.PathEscape(roomID)+"/join",
+ bob.AccessToken, map[string]any{}), http.StatusOK)
+
+ aliceKey := callMemberPath(roomID, "_"+alice.UserID+"_A_m.call")
+ ts.scheduleDelayed(aliceKey, 5000, alice, map[string]any{})
+
+ // Alice refreshing her own membership leaves the delayed leave alone.
+ expectStatus(t, ts.put(aliceKey, alice.AccessToken, map[string]any{"application": "m.call"}), http.StatusOK)
+ // Bob, an ordinary member, may put call state of his own.
+ expectStatus(t, ts.put(callMemberPath(roomID, "_"+bob.UserID+"_B_m.call"), bob.AccessToken,
+ map[string]any{"application": "m.call"}), http.StatusOK)
+ // The cancellation runs off the emit stage, so give it a moment.
+ time.Sleep(100 * time.Millisecond)
+ if l := ts.delayedList(alice); len(l.Scheduled) != 1 {
+ t.Fatalf("alice's own update cancelled her delayed event: %+v", l)
+ }
+
+ // Bob writing the same piece of state cancels it.
+ expectStatus(t, ts.put(aliceKey, bob.AccessToken, map[string]any{}), http.StatusOK)
+ deadline := time.Now().Add(5 * time.Second)
+ for {
+ l := ts.delayedList(alice)
+ if len(l.Scheduled) == 0 {
+ if len(l.Finalised) != 1 || l.Finalised[0].Outcome != "cancel" {
+ t.Errorf("finalised = %+v", l.Finalised)
+ }
+ break
+ }
+ if time.Now().After(deadline) {
+ t.Fatal("the overwritten delayed event was never cancelled")
+ }
+ time.Sleep(10 * time.Millisecond)
+ }
+}
+
+func TestDelayedEventLimits(t *testing.T) {
+ ts := newTestServer(t, nil)
+ alice := ts.register("alice", "correct-horse-battery-staple")
+ roomID := ts.createRoom(alice, map[string]any{"preset": "public_chat"})
+ topic := "/_matrix/client/v3/rooms/" + url.PathEscape(roomID) + "/state/m.room.topic"
+
+ // Longer than the server allows: the MSC's own code and the maximum
+ // come with the error.
+ rec := ts.put(topic+"?org.matrix.msc4140.delay=7200000", alice.AccessToken, map[string]any{"topic": "x"})
+ expectError(t, rec, http.StatusBadRequest, httputil.ErrCodeUnknown)
+ body := jsonBody(t, rec)
+ if body["org.matrix.msc4140.errcode"] != "M_MAX_DELAY_EXCEEDED" || body["org.matrix.msc4140.max_delay"] != float64(3600000) {
+ t.Errorf("error = %s", rec.Body.String())
+ }
+
+ expectError(t, ts.put(topic+"?org.matrix.msc4140.delay=soon", alice.AccessToken, map[string]any{"topic": "x"}),
+ http.StatusBadRequest, httputil.ErrCodeInvalidParam)
+ expectError(t, ts.put(topic+"?org.matrix.msc4140.parent_delay_id=x", alice.AccessToken, map[string]any{"topic": "x"}),
+ http.StatusBadRequest, httputil.ErrCodeInvalidParam)
+
+ // Five at once is the test server's limit.
+ for range 5 {
+ ts.scheduleDelayed(topic, 1000, alice, map[string]any{"topic": "x"})
+ }
+ rec = ts.put(topic+"?org.matrix.msc4140.delay=1000", alice.AccessToken, map[string]any{"topic": "x"})
+ expectError(t, rec, http.StatusBadRequest, "M_LIMIT_EXCEEDED")
+
+ // Nobody else can read them.
+ bob := ts.register("bob", "correct-horse-battery-staple")
+ if l := ts.delayedList(bob); len(l.Scheduled) != 0 {
+ t.Errorf("bob sees alice's delayed events: %+v", l)
+ }
+ expectError(t, ts.get(delayedEventsPath, ""), http.StatusUnauthorized, "M_MISSING_TOKEN")
+}
+
+// TestDelayedEventsDisabled: without the service the parameter is refused,
+// rather than the event being sent at once, and the endpoints are absent.
+func TestDelayedEventsDisabled(t *testing.T) {
+ ts := newTestServer(t, func(o *clientapi.Options) { o.DelayedEvents = nil })
+ alice := ts.register("alice", "correct-horse-battery-staple")
+ roomID := ts.createRoom(alice, map[string]any{"preset": "public_chat"})
+ path := callMemberPath(roomID, "_"+alice.UserID+"_A_m.call")
+
+ expectError(t, ts.put(path+"?org.matrix.msc4140.delay=1000", alice.AccessToken, map[string]any{}),
+ http.StatusBadRequest, httputil.ErrCodeInvalidParam)
+ expectError(t, ts.get(path, alice.AccessToken), http.StatusNotFound, httputil.ErrCodeNotFound)
+ expectError(t, ts.get(delayedEventsPath, alice.AccessToken), http.StatusNotFound, httputil.ErrCodeUnrecognized)
+}
diff --git a/server/internal/clientapi/openid.go b/server/internal/clientapi/openid.go
new file mode 100644
index 0000000..b2497a6
--- /dev/null
+++ b/server/internal/clientapi/openid.go
@@ -0,0 +1,106 @@
+package clientapi
+
+import (
+ "crypto/rand"
+ "encoding/base64"
+ "errors"
+ "fmt"
+ "net/http"
+ "time"
+
+ "github.com/dariuszjaniak/warp/server/internal/httputil"
+ "github.com/dariuszjaniak/warp/server/internal/storage"
+ "github.com/dariuszjaniak/warp/server/internal/userapi"
+)
+
+// openIDTokenLifetime is how long an OpenID token is good for. The LiveKit
+// JWT service checks it once, when a call is joined, so an hour is plenty.
+const openIDTokenLifetime = time.Hour
+
+// openIDTokenPrefix marks OpenID tokens the way the userapi marks access
+// tokens, so one handed to the wrong endpoint is recognisable.
+const openIDTokenPrefix = "warp_oi_"
+
+// openIDUserInfoPath is where a third party exchanges an OpenID token for
+// the user's ID. It is a federation endpoint by name, but needs neither
+// federation nor a signing key: Synapse serves it as a resource of its own
+// for the same reason. It is registered here so that a server with
+// federation off can still let its users into a group call.
+const openIDUserInfoPath = "/_matrix/federation/v1/openid/userinfo"
+
+// registerOpenIDRoutes adds the OpenID endpoints, if there is a store for
+// the tokens.
+func (c *ClientAPI) registerOpenIDRoutes(r *httputil.Router) {
+ if c.openID == nil {
+ return
+ }
+ c.authed(r, "POST "+v3+"/user/{userID}/openid/request_token", c.limitDefault, c.handleRequestOpenIDToken)
+ c.public(r, "GET "+openIDUserInfoPath, c.limitDefault, c.handleOpenIDUserInfo)
+}
+
+type openIDTokenResponse struct {
+ AccessToken string `json:"access_token"`
+ TokenType string `json:"token_type"`
+ MatrixServerName string `json:"matrix_server_name"`
+ ExpiresIn int64 `json:"expires_in"`
+}
+
+// handleRequestOpenIDToken implements
+// POST /_matrix/client/v3/user/{userID}/openid/request_token.
+//
+// The token proves who the user is to whoever they hand it to, and nothing
+// else: it is not an access token and opens no other endpoint.
+func (c *ClientAPI) handleRequestOpenIDToken(w http.ResponseWriter, r *http.Request, auth userapi.Authenticated) error {
+ if r.PathValue("userID") != auth.UserID {
+ return httputil.Forbidden("You cannot request an OpenID token for another user")
+ }
+ // The body is an empty object; anything in it is ignored.
+ var ignored map[string]any
+ if _, err := httputil.DecodeOptionalJSON(w, r, &ignored); err != nil {
+ return err
+ }
+
+ raw := make([]byte, 32)
+ if _, err := rand.Read(raw); err != nil {
+ return fmt.Errorf("generate OpenID token: %w", err)
+ }
+ token := openIDTokenPrefix + base64.RawURLEncoding.EncodeToString(raw)
+ now := c.now()
+ // Expired tokens go as new ones are made; nothing else reads them.
+ if err := c.openID.PurgeOpenIDTokens(r.Context(), now.UnixMilli()); err != nil {
+ c.log.Warn("could not purge expired OpenID tokens", "error", err)
+ }
+ if err := c.openID.InsertOpenIDToken(r.Context(), userapi.HashToken(token), auth.UserID,
+ now.Add(openIDTokenLifetime).UnixMilli()); err != nil {
+ return err
+ }
+ httputil.WriteJSON(w, http.StatusOK, openIDTokenResponse{
+ AccessToken: token,
+ TokenType: "Bearer",
+ MatrixServerName: c.serverName,
+ ExpiresIn: int64(openIDTokenLifetime.Seconds()),
+ })
+ return nil
+}
+
+type openIDUserInfoResponse struct {
+ Sub string `json:"sub"`
+}
+
+// handleOpenIDUserInfo implements
+// GET /_matrix/federation/v1/openid/userinfo.
+func (c *ClientAPI) handleOpenIDUserInfo(w http.ResponseWriter, r *http.Request) error {
+ token := r.URL.Query().Get("access_token")
+ if token == "" {
+ return httputil.MissingToken()
+ }
+ userID, err := c.openID.OpenIDTokenUser(r.Context(), userapi.HashToken(token), c.now().UnixMilli())
+ if errors.Is(err, storage.ErrNoSuchOpenIDToken) {
+ return httputil.UnknownToken(false)
+ }
+ if err != nil {
+ return err
+ }
+ httputil.WriteJSON(w, http.StatusOK, openIDUserInfoResponse{Sub: userID})
+ return nil
+}
diff --git a/server/internal/clientapi/openid_test.go b/server/internal/clientapi/openid_test.go
new file mode 100644
index 0000000..5ec3ac5
--- /dev/null
+++ b/server/internal/clientapi/openid_test.go
@@ -0,0 +1,61 @@
+package clientapi_test
+
+import (
+ "net/http"
+ "net/url"
+ "testing"
+
+ "github.com/dariuszjaniak/warp/server/internal/httputil"
+)
+
+// TestOpenIDToken is the exchange the LiveKit JWT service relies on: a user
+// asks for a token, hands it over, and the service asks this server who it
+// belongs to.
+func TestOpenIDToken(t *testing.T) {
+ ts := newTestServer(t, nil)
+ alice := ts.register("alice", "correct-horse-battery-staple")
+ bob := ts.register("bob", "correct-horse-battery-staple")
+ path := "/_matrix/client/v3/user/" + url.PathEscape(alice.UserID) + "/openid/request_token"
+
+ rec := ts.post(path, alice.AccessToken, map[string]any{})
+ expectStatus(t, rec, http.StatusOK)
+ body := jsonBody(t, rec)
+ token, _ := body["access_token"].(string)
+ if token == "" || body["token_type"] != "Bearer" || body["matrix_server_name"] != testServerName ||
+ body["expires_in"] != float64(3600) {
+ t.Fatalf("request_token = %s", rec.Body.String())
+ }
+
+ // The token is no access token.
+ expectError(t, ts.get("/_matrix/client/v3/account/whoami", token), http.StatusUnauthorized, httputil.ErrCodeUnknownToken)
+
+ rec = ts.get("/_matrix/federation/v1/openid/userinfo?access_token="+url.QueryEscape(token), "")
+ expectStatus(t, rec, http.StatusOK)
+ if got := jsonBody(t, rec)["sub"]; got != alice.UserID {
+ t.Errorf("sub = %v, want %s", got, alice.UserID)
+ }
+
+ expectError(t, ts.get("/_matrix/federation/v1/openid/userinfo?access_token=nope", ""),
+ http.StatusUnauthorized, httputil.ErrCodeUnknownToken)
+ expectError(t, ts.get("/_matrix/federation/v1/openid/userinfo", ""),
+ http.StatusUnauthorized, "M_MISSING_TOKEN")
+
+ // Nobody gets a token in someone else's name.
+ expectError(t, ts.post(path, bob.AccessToken, map[string]any{}), http.StatusForbidden, httputil.ErrCodeForbidden)
+}
+
+// TestOpenIDUserInfoIgnoresFederationSwitch: the userinfo endpoint answers
+// with federation off, so a server that does not federate can still let its
+// users into group calls.
+func TestOpenIDUserInfoIgnoresFederationSwitch(t *testing.T) {
+ ts := newTestServer(t, nil)
+ if ts.settings.FederationEnabled() {
+ t.Fatal("the test server is expected to start with federation off")
+ }
+ alice := ts.register("alice", "correct-horse-battery-staple")
+ rec := ts.post("/_matrix/client/v3/user/"+url.PathEscape(alice.UserID)+"/openid/request_token",
+ alice.AccessToken, nil)
+ expectStatus(t, rec, http.StatusOK)
+ token, _ := jsonBody(t, rec)["access_token"].(string)
+ expectStatus(t, ts.get("/_matrix/federation/v1/openid/userinfo?access_token="+url.QueryEscape(token), ""), http.StatusOK)
+}
diff --git a/server/internal/clientapi/rooms.go b/server/internal/clientapi/rooms.go
index 26c6b2f..72b3ec8 100644
--- a/server/internal/clientapi/rooms.go
+++ b/server/internal/clientapi/rooms.go
@@ -6,6 +6,7 @@ import (
"net/http"
"strings"
+ "github.com/dariuszjaniak/warp/server/internal/delayedevents"
"github.com/dariuszjaniak/warp/server/internal/httputil"
"github.com/dariuszjaniak/warp/server/internal/matrix/id"
"github.com/dariuszjaniak/warp/server/internal/roomserver"
@@ -155,10 +156,20 @@ func (c *ClientAPI) handleSendEvent(w http.ResponseWriter, r *http.Request, auth
return httputil.MissingParam("A transaction ID is required")
}
+ delay, delayed, err := c.delayRequested(r)
+ if err != nil {
+ return err
+ }
var content sendEventRequest
if err := httputil.DecodeJSON(w, r, &content); err != nil {
return err
}
+ if delayed {
+ return c.scheduleDelayed(w, r, delayedevents.ScheduleRequest{
+ UserID: auth.UserID, DeviceID: auth.DeviceID, RoomID: roomID,
+ Type: eventType, Content: content, Delay: delay,
+ })
+ }
eventID, err := c.rooms.SendEvent(r.Context(), roomserver.SendEventRequest{
RoomID: roomID,
@@ -191,10 +202,20 @@ func (c *ClientAPI) handlePutState(w http.ResponseWriter, r *http.Request, auth
}
stateKey := r.PathValue("stateKey")
+ delay, delayed, err := c.delayRequested(r)
+ if err != nil {
+ return err
+ }
var content sendEventRequest
if err := httputil.DecodeJSON(w, r, &content); err != nil {
return err
}
+ if delayed {
+ return c.scheduleDelayed(w, r, delayedevents.ScheduleRequest{
+ UserID: auth.UserID, DeviceID: auth.DeviceID, RoomID: roomID,
+ Type: eventType, StateKey: &stateKey, Content: content, Delay: delay,
+ })
+ }
eventID, err := c.rooms.SendEvent(r.Context(), roomserver.SendEventRequest{
RoomID: roomID,
diff --git a/server/internal/clientapi/rtc.go b/server/internal/clientapi/rtc.go
new file mode 100644
index 0000000..d428b20
--- /dev/null
+++ b/server/internal/clientapi/rtc.go
@@ -0,0 +1,51 @@
+package clientapi
+
+import (
+ "net/http"
+ "strings"
+
+ "github.com/dariuszjaniak/warp/server/internal/httputil"
+ "github.com/dariuszjaniak/warp/server/internal/userapi"
+)
+
+// rtcTransportsPath is MSC4143's endpoint for the group call backends a
+// client may use: the same list /.well-known/matrix/client carries as
+// org.matrix.msc4143.rtc_foci, behind authentication and without needing
+// the discovery document.
+const rtcTransportsPath = "/_matrix/client/unstable/org.matrix.msc4143/rtc/transports"
+
+// RTCTransport is one group call backend: a LiveKit SFU, named by the JWT
+// service that hands out tokens for it.
+type RTCTransport struct {
+ Type string `json:"type"`
+ LiveKitServiceURL string `json:"livekit_service_url"`
+}
+
+// RTCTransports lists the configured group call backends, or nil for none.
+func RTCTransports(liveKitServiceURL string) []RTCTransport {
+ if liveKitServiceURL == "" {
+ return nil
+ }
+ return []RTCTransport{{Type: "livekit", LiveKitServiceURL: strings.TrimRight(liveKitServiceURL, "/")}}
+}
+
+type rtcTransportsResponse struct {
+ Transports []RTCTransport `json:"rtc_transports"`
+}
+
+// registerRTCRoutes adds the MSC4143 endpoint, if a backend is configured.
+// Without one it is left unregistered: the 404 is how a client learns there
+// is none.
+func (c *ClientAPI) registerRTCRoutes(r *httputil.Router) {
+ if len(c.rtcTransports) == 0 {
+ return
+ }
+ c.authed(r, "GET "+rtcTransportsPath, c.limitDefault, c.handleRTCTransports)
+}
+
+// handleRTCTransports implements
+// GET /_matrix/client/unstable/org.matrix.msc4143/rtc/transports.
+func (c *ClientAPI) handleRTCTransports(w http.ResponseWriter, _ *http.Request, _ userapi.Authenticated) error {
+ httputil.WriteJSON(w, http.StatusOK, rtcTransportsResponse{Transports: c.rtcTransports})
+ return nil
+}
diff --git a/server/internal/clientapi/voip_test.go b/server/internal/clientapi/voip_test.go
index 5ad3e2c..387fd5c 100644
--- a/server/internal/clientapi/voip_test.go
+++ b/server/internal/clientapi/voip_test.go
@@ -87,3 +87,27 @@ func TestTurnServerStatic(t *testing.T) {
t.Errorf("response = %+v", got)
}
}
+
+func TestRTCTransports(t *testing.T) {
+ const path = "/_matrix/client/unstable/org.matrix.msc4143/rtc/transports"
+ ts := newTestServer(t, func(o *clientapi.Options) {
+ o.RTC = config.RTC{LiveKitServiceURL: "https://livekit-jwt.warp.example/"}
+ })
+ acc := ts.register("alice", "correct horse battery")
+ expectError(t, ts.get(path, ""), http.StatusUnauthorized, "M_MISSING_TOKEN")
+
+ rec := ts.get(path, acc.AccessToken)
+ expectStatus(t, rec, http.StatusOK)
+ got := decode[struct {
+ Transports []map[string]string `json:"rtc_transports"`
+ }](t, rec)
+ if len(got.Transports) != 1 || got.Transports[0]["type"] != "livekit" ||
+ got.Transports[0]["livekit_service_url"] != "https://livekit-jwt.warp.example" {
+ t.Errorf("rtc_transports = %v", got.Transports)
+ }
+
+ // Without a backend there is no endpoint, which is how a client knows.
+ ts = newTestServer(t, nil)
+ acc = ts.register("alice", "correct horse battery")
+ expectError(t, ts.get(path, acc.AccessToken), http.StatusNotFound, "M_UNRECOGNIZED")
+}
diff --git a/server/internal/config/config.go b/server/internal/config/config.go
index 75a7ce2..61f81d9 100644
--- a/server/internal/config/config.go
+++ b/server/internal/config/config.go
@@ -182,6 +182,39 @@ type ClientAPI struct {
// TURN is what GET /voip/turnServer hands clients for voice and video
// calls.
TURN TURN `yaml:"turn"`
+
+ // RTC is the media backend for group calls (MatrixRTC), advertised in
+ // /.well-known/matrix/client.
+ RTC RTC `yaml:"rtc"`
+
+ // DelayedEvents bounds the delayed events (MSC4140) a client may
+ // schedule. Group calls depend on them: a member's "I left" is
+ // scheduled in advance, so a client that disappears still leaves.
+ DelayedEvents DelayedEvents `yaml:"delayed_events"`
+}
+
+// RTC configures group calls. Warp does not carry call media itself:
+// MatrixRTC clients publish to a LiveKit SFU, and get the token for it from
+// a LiveKit JWT service (lk-jwt-service) in exchange for an OpenID token
+// from this server. With no service URL no focus is advertised, and clients
+// that need one cannot start a group call.
+type RTC struct {
+ // LiveKitServiceURL is the base URL of the LiveKit JWT service, e.g.
+ // "https://livekit-jwt.example.org". It is advertised as an
+ // org.matrix.msc4143.rtc_foci entry of type livekit.
+ LiveKitServiceURL string `yaml:"livekit_service_url" env:"WARP_RTC_LIVEKIT_SERVICE_URL"`
+}
+
+// DelayedEvents configures MSC4140 delayed events.
+type DelayedEvents struct {
+ // MaxDelay is the longest delay a client may ask for. Zero turns
+ // delayed events off: the delay parameter is refused and the feature
+ // is not advertised.
+ MaxDelay time.Duration `yaml:"max_delay" env:"WARP_DELAYED_EVENTS_MAX_DELAY"`
+
+ // MaxPerUser caps how many delayed events one user may have scheduled
+ // at once.
+ MaxPerUser int `yaml:"max_per_user" env:"WARP_DELAYED_EVENTS_MAX_PER_USER"`
}
// TURN configures the relay servers clients use to set up calls. With no
@@ -531,6 +564,10 @@ func Default() *Config {
Media: RateLimitBucket{Burst: 100, Interval: 20 * time.Millisecond},
},
TURN: TURN{UserLifetime: time.Hour},
+ // A MatrixRTC client restarts its delayed leave every few
+ // seconds; a day leaves room for other uses without letting a
+ // forgotten event linger for weeks.
+ DelayedEvents: DelayedEvents{MaxDelay: 24 * time.Hour, MaxPerUser: 100},
},
Sync: Sync{
Presence: false,
@@ -714,6 +751,17 @@ func (c *Config) Validate() error {
if c.ClientAPI.TURN.SharedSecret != "" && c.ClientAPI.TURN.UserLifetime <= 0 {
bad("client_api.turn.user_lifetime: must be positive")
}
+ if raw := c.ClientAPI.RTC.LiveKitServiceURL; raw != "" {
+ if u, err := url.Parse(raw); err != nil || (u.Scheme != "http" && u.Scheme != "https") || u.Host == "" {
+ bad("client_api.rtc.livekit_service_url: must be an http(s) URL, got %q", raw)
+ }
+ }
+ if c.ClientAPI.DelayedEvents.MaxDelay < 0 {
+ bad("client_api.delayed_events.max_delay: must not be negative")
+ }
+ if c.ClientAPI.DelayedEvents.MaxDelay > 0 && c.ClientAPI.DelayedEvents.MaxPerUser < 1 {
+ bad("client_api.delayed_events.max_per_user: must be at least 1")
+ }
if c.Sync.TypingTimeout <= 0 {
bad("sync.typing_timeout: must be positive")
}
diff --git a/server/internal/delayedevents/delayedevents.go b/server/internal/delayedevents/delayedevents.go
new file mode 100644
index 0000000..5adf392
--- /dev/null
+++ b/server/internal/delayedevents/delayedevents.go
@@ -0,0 +1,463 @@
+// Package delayedevents implements MSC4140 delayed events: an event a client
+// hands the server now, to be sent later on the client's behalf unless the
+// client pushes it back or cancels it first.
+//
+// It is what makes MatrixRTC group calls robust. A member joining a call
+// sends its call.member state, and schedules an empty one to replace it a
+// few seconds out; while it is in the call it keeps restarting that delay.
+// A client that crashes, loses its network or is closed stops restarting
+// it, and the server sends the "left" for it.
+//
+// Events are kept in the database, so a restart of the server does not
+// lose them, and are sent by a single scheduler goroutine that sleeps until
+// the next one is due.
+package delayedevents
+
+import (
+ "context"
+ "crypto/rand"
+ "encoding/base64"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "log/slog"
+ "sync"
+ "time"
+
+ "github.com/dariuszjaniak/warp/server/internal/roomserver"
+ "github.com/dariuszjaniak/warp/server/internal/storage"
+)
+
+// Outcomes and reasons a finalised event is reported with, as MSC4140
+// names them.
+const (
+ OutcomeSend = "send"
+ OutcomeCancel = "cancel"
+
+ ReasonDelay = "delay"
+ ReasonAction = "action"
+ ReasonError = "error"
+)
+
+var (
+ // ErrDisabled is returned when delayed events are switched off.
+ ErrDisabled = errors.New("delayed events are disabled")
+
+ // ErrMaxDelayExceeded is returned for a delay longer than the
+ // configured maximum.
+ ErrMaxDelayExceeded = errors.New("the requested delay exceeds the allowed maximum")
+
+ // ErrTooMany is returned when the user already has as many delayed
+ // events scheduled as they may.
+ ErrTooMany = errors.New("too many delayed events are scheduled")
+
+ // ErrNoSuchDelayedEvent is returned for a delay ID that is unknown or
+ // no longer scheduled.
+ ErrNoSuchDelayedEvent = storage.ErrNoSuchDelayedEvent
+)
+
+// finalisedRetention is how long a sent or cancelled event is kept for a
+// client to look up.
+const finalisedRetention = 24 * time.Hour
+
+// idleCheck is how long the scheduler sleeps with nothing scheduled. It
+// also paces the purge of finalised events.
+const idleCheck = time.Hour
+
+// dueBatch is how many due events are read at once.
+const dueBatch = 100
+
+// Sender sends an event into a room as a user: the roomserver.
+type Sender interface {
+ SendEvent(ctx context.Context, req roomserver.SendEventRequest) (string, error)
+}
+
+// Options configure the service.
+type Options struct {
+ Store storage.DelayedEventStore
+ Rooms Sender
+
+ // MaxDelay is the longest delay allowed. Zero disables the service.
+ MaxDelay time.Duration
+
+ // MaxPerUser caps the events one user may have scheduled.
+ MaxPerUser int
+
+ // DescribeError turns the error a send failed with into the Matrix
+ // error a client is shown. Nil reports every failure as M_UNKNOWN.
+ DescribeError func(error) json.RawMessage
+
+ Logger *slog.Logger
+ Now func() time.Time
+}
+
+// Service schedules and sends delayed events.
+type Service struct {
+ store storage.DelayedEventStore
+ rooms Sender
+ maxDelay time.Duration
+ maxPerUser int
+ describeError func(error) json.RawMessage
+ log *slog.Logger
+ now func() time.Time
+
+ // mu serialises everything that finalises or moves an event, so the
+ // scheduler sending an event and its owner restarting or cancelling
+ // it cannot both win.
+ mu sync.Mutex
+
+ wake chan struct{}
+ stop chan struct{}
+ done chan struct{}
+ started bool
+ closing sync.Once
+
+ // cancels tracks the background cancellations the emit stage starts,
+ // so Close can wait for them.
+ cancels sync.WaitGroup
+}
+
+// New builds the service. It does not start the scheduler.
+func New(opts Options) (*Service, error) {
+ if opts.Store == nil {
+ return nil, errors.New("delayedevents: Store must not be nil")
+ }
+ if opts.Rooms == nil {
+ return nil, errors.New("delayedevents: Rooms must not be nil")
+ }
+ if opts.Logger == nil {
+ opts.Logger = slog.Default()
+ }
+ if opts.Now == nil {
+ opts.Now = time.Now
+ }
+ if opts.DescribeError == nil {
+ opts.DescribeError = func(err error) json.RawMessage {
+ raw, _ := json.Marshal(map[string]string{"errcode": "M_UNKNOWN", "error": err.Error()})
+ return raw
+ }
+ }
+ return &Service{
+ store: opts.Store,
+ rooms: opts.Rooms,
+ maxDelay: opts.MaxDelay,
+ maxPerUser: opts.MaxPerUser,
+ describeError: opts.DescribeError,
+ log: opts.Logger.With("component", "delayedevents"),
+ now: opts.Now,
+ wake: make(chan struct{}, 1),
+ stop: make(chan struct{}),
+ done: make(chan struct{}),
+ }, nil
+}
+
+// Enabled reports whether clients may schedule delayed events.
+func (s *Service) Enabled() bool { return s != nil && s.maxDelay > 0 }
+
+// MaxDelay is the longest delay a client may ask for.
+func (s *Service) MaxDelay() time.Duration { return s.maxDelay }
+
+// ScheduleRequest is an event to send later.
+type ScheduleRequest struct {
+ UserID string
+ DeviceID string
+ RoomID string
+ Type string
+
+ // StateKey makes this a state event when it is non-nil.
+ StateKey *string
+ Content json.RawMessage
+ Delay time.Duration
+}
+
+// Schedule stores an event to be sent after req.Delay and returns its
+// delay ID.
+func (s *Service) Schedule(ctx context.Context, req ScheduleRequest) (string, error) {
+ if !s.Enabled() {
+ return "", ErrDisabled
+ }
+ if req.Delay < 0 {
+ return "", fmt.Errorf("%w: the delay must not be negative", roomserver.ErrInvalidContent)
+ }
+ if req.Delay > s.maxDelay {
+ return "", ErrMaxDelayExceeded
+ }
+ n, err := s.store.CountScheduledDelayedEvents(ctx, req.UserID)
+ if err != nil {
+ return "", err
+ }
+ if n >= s.maxPerUser {
+ return "", ErrTooMany
+ }
+ delayID, err := newDelayID()
+ if err != nil {
+ return "", err
+ }
+ nowMS := s.now().UnixMilli()
+ delayMS := req.Delay.Milliseconds()
+ if err := s.store.InsertDelayedEvent(ctx, storage.DelayedEvent{
+ DelayID: delayID,
+ UserID: req.UserID,
+ DeviceID: req.DeviceID,
+ RoomID: req.RoomID,
+ Type: req.Type,
+ StateKey: req.StateKey,
+ Content: req.Content,
+ DelayMS: delayMS,
+ RunningSinceMS: nowMS,
+ SendAtMS: nowMS + delayMS,
+ }); err != nil {
+ return "", err
+ }
+ s.poke()
+ return delayID, nil
+}
+
+// Get returns one of a user's delayed events, scheduled or finalised.
+func (s *Service) Get(ctx context.Context, userID, delayID string) (storage.DelayedEvent, error) {
+ return s.store.DelayedEvent(ctx, userID, delayID)
+}
+
+// List returns a user's delayed events.
+func (s *Service) List(ctx context.Context, userID string) ([]storage.DelayedEvent, error) {
+ return s.store.DelayedEvents(ctx, userID)
+}
+
+// Restart starts a scheduled event's delay again from now.
+//
+// The action endpoints may be called without an access token: the delay
+// ID is the capability. That is what lets a MatrixRTC client hand its
+// delayed leave to a service that keeps it alive on its behalf.
+func (s *Service) Restart(ctx context.Context, delayID string) error {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ if err := s.store.RestartDelayedEvent(ctx, delayID, s.now().UnixMilli()); err != nil {
+ return err
+ }
+ s.poke()
+ return nil
+}
+
+// Cancel drops a scheduled event without sending it.
+func (s *Service) Cancel(ctx context.Context, delayID string) error {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ e, err := s.store.ScheduledDelayedEvent(ctx, delayID)
+ if err != nil {
+ return err
+ }
+ return s.finalise(ctx, e, OutcomeCancel, ReasonAction, "", nil)
+}
+
+// Send sends a scheduled event now, without waiting for its delay. The
+// error of a failed send is returned as well as recorded.
+func (s *Service) Send(ctx context.Context, delayID string) error {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ e, err := s.store.ScheduledDelayedEvent(ctx, delayID)
+ if err != nil {
+ return err
+ }
+ return s.sendLocked(ctx, e, ReasonAction)
+}
+
+// sendLocked sends e and records the outcome. s.mu must be held.
+func (s *Service) sendLocked(ctx context.Context, e storage.DelayedEvent, reason string) error {
+ eventID, sendErr := s.rooms.SendEvent(ctx, roomserver.SendEventRequest{
+ RoomID: e.RoomID,
+ Sender: e.UserID,
+ Type: e.Type,
+ StateKey: e.StateKey,
+ Content: e.Content,
+ })
+ if sendErr != nil {
+ if err := s.finalise(ctx, e, OutcomeSend, ReasonError, "", s.describeError(sendErr)); err != nil {
+ return err
+ }
+ return sendErr
+ }
+ return s.finalise(ctx, e, OutcomeSend, reason, eventID, nil)
+}
+
+func (s *Service) finalise(ctx context.Context, e storage.DelayedEvent, outcome, reason, eventID string, errBody json.RawMessage) error {
+ e.FinalisedAtMS = s.now().UnixMilli()
+ e.Outcome = outcome
+ e.Reason = reason
+ e.EventID = eventID
+ e.Error = errBody
+ return s.store.FinaliseDelayedEvent(ctx, e)
+}
+
+// EventPersisted implements roomserver.Notifier.
+//
+// When someone else sets a piece of state, the delayed state events that
+// would overwrite it are cancelled: a delayed event stands for "put this
+// back if I stop answering", and must not clobber a change another user
+// made in the meantime. The owner's own updates leave them alone, which is
+// what lets a call member refresh its membership while its delayed leave
+// keeps running.
+//
+// The emit stage must not block, so the lookup runs in the background.
+func (s *Service) EventPersisted(_ context.Context, n roomserver.Notification) {
+ if !s.Enabled() || n.Event == nil || !n.StateChanged {
+ return
+ }
+ key, isState := n.Event.StateKey()
+ if !isState {
+ return
+ }
+ roomID, eventType, sender := n.RoomID, n.Event.Type(), n.Event.Sender()
+ select {
+ case <-s.stop:
+ return
+ default:
+ }
+ s.cancels.Add(1)
+ go func() {
+ defer s.cancels.Done()
+ ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+ defer cancel()
+ s.cancelOverwritten(ctx, roomID, eventType, key, sender)
+ }()
+}
+
+func (s *Service) cancelOverwritten(ctx context.Context, roomID, eventType, stateKey, sender string) {
+ pending, err := s.store.ScheduledDelayedStateEvents(ctx, roomID, eventType, stateKey)
+ if err != nil {
+ s.log.Warn("could not look up delayed state events", "room_id", roomID, "error", err)
+ return
+ }
+ for _, e := range pending {
+ if e.UserID == sender {
+ continue
+ }
+ s.mu.Lock()
+ err := s.finalise(ctx, e, OutcomeCancel, ReasonAction, "", nil)
+ s.mu.Unlock()
+ if err != nil && !errors.Is(err, storage.ErrNoSuchDelayedEvent) {
+ s.log.Warn("could not cancel an overwritten delayed event", "delay_id", e.DelayID, "error", err)
+ }
+ }
+}
+
+// Start runs the scheduler until ctx is cancelled or Close is called.
+func (s *Service) Start(ctx context.Context) {
+ if !s.Enabled() || s.started {
+ return
+ }
+ s.started = true
+ go s.run(ctx)
+}
+
+// Close stops the scheduler and waits for it.
+func (s *Service) Close() {
+ s.closing.Do(func() { close(s.stop) })
+ if s.started {
+ <-s.done
+ }
+ s.cancels.Wait()
+}
+
+// poke wakes the scheduler to look at the schedule again.
+func (s *Service) poke() {
+ select {
+ case s.wake <- struct{}{}:
+ default:
+ }
+}
+
+func (s *Service) run(ctx context.Context) {
+ defer close(s.done)
+ timer := time.NewTimer(0)
+ defer timer.Stop()
+ lastPurge := time.Time{}
+
+ for {
+ select {
+ case <-ctx.Done():
+ return
+ case <-s.stop:
+ return
+ case <-timer.C:
+ case <-s.wake:
+ }
+
+ s.SendDue(ctx)
+ if now := s.now(); now.Sub(lastPurge) >= idleCheck {
+ lastPurge = now
+ if err := s.store.PurgeFinalisedDelayedEvents(ctx, now.Add(-finalisedRetention).UnixMilli()); err != nil {
+ s.log.Warn("could not purge finalised delayed events", "error", err)
+ }
+ }
+
+ wait := idleCheck
+ next, ok, err := s.store.NextDelayedEventDue(ctx)
+ switch {
+ case err != nil:
+ if ctx.Err() != nil {
+ return
+ }
+ s.log.Warn("could not read the delayed event schedule", "error", err)
+ wait = time.Second
+ case ok:
+ wait = min(max(time.Duration(next-s.now().UnixMilli())*time.Millisecond, 0), idleCheck)
+ }
+ if !timer.Stop() {
+ select {
+ case <-timer.C:
+ default:
+ }
+ }
+ timer.Reset(wait)
+ }
+}
+
+// SendDue sends every event whose delay has run out. The scheduler calls
+// it whenever the next event comes due; tests call it after moving the
+// clock.
+func (s *Service) SendDue(ctx context.Context) {
+ for {
+ due, err := s.store.DueDelayedEvents(ctx, s.now().UnixMilli(), dueBatch)
+ if err != nil {
+ if ctx.Err() == nil {
+ s.log.Warn("could not read due delayed events", "error", err)
+ }
+ return
+ }
+ for _, e := range due {
+ s.sendIfStillDue(ctx, e.DelayID)
+ }
+ if len(due) < dueBatch {
+ return
+ }
+ }
+}
+
+// sendIfStillDue re-reads an event under the lock, because its owner may
+// have restarted or cancelled it since the batch was read.
+func (s *Service) sendIfStillDue(ctx context.Context, delayID string) {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ e, err := s.store.ScheduledDelayedEvent(ctx, delayID)
+ if err != nil {
+ return
+ }
+ if e.SendAtMS > s.now().UnixMilli() {
+ return
+ }
+ if err := s.sendLocked(ctx, e, ReasonDelay); err != nil {
+ s.log.Info("a delayed event could not be sent",
+ "delay_id", e.DelayID, "room_id", e.RoomID, "type", e.Type, "error", err)
+ }
+}
+
+// newDelayID makes a delay ID. It is a capability — anyone holding it may
+// restart, cancel or send the event — so it is as long as an access token's
+// random part.
+func newDelayID() (string, error) {
+ raw := make([]byte, 24)
+ if _, err := rand.Read(raw); err != nil {
+ return "", fmt.Errorf("generate delay ID: %w", err)
+ }
+ return "warp_de_" + base64.RawURLEncoding.EncodeToString(raw), nil
+}
diff --git a/server/internal/httputil/errors.go b/server/internal/httputil/errors.go
index 9e49e09..3cf974d 100644
--- a/server/internal/httputil/errors.go
+++ b/server/internal/httputil/errors.go
@@ -81,6 +81,11 @@ type MatrixError struct {
RemoteStatus int `json:"status,omitempty"`
RemoteBody string `json:"body,omitempty"`
+ // DelayedErrCode and MaxDelayMS accompany a delayed event whose delay
+ // is longer than the server allows (MSC4140).
+ DelayedErrCode string `json:"org.matrix.msc4140.errcode,omitempty"`
+ MaxDelayMS int64 `json:"org.matrix.msc4140.max_delay,omitempty"`
+
// status is the HTTP status to send; not part of the JSON body.
status int
}
diff --git a/server/internal/roomserver/create.go b/server/internal/roomserver/create.go
index e85c65a..4a213c9 100644
--- a/server/internal/roomserver/create.go
+++ b/server/internal/roomserver/create.go
@@ -432,6 +432,10 @@ func (r *API) powerLevelsContent(req CreateRoomRequest, preset string, rules roo
"m.room.avatar": auth.DefaultStateDefault,
TypeEncryption: 100,
"m.room.server_acl": 100,
+ // Joining a group call is a state event of the member's own
+ // (MatrixRTC, MSC3401), so anyone who may talk in the room
+ // may also join its call. Starting one is not a separate act.
+ TypeCallMember: 0,
},
"notifications": map[string]any{"room": auth.DefaultNotification},
}
@@ -538,3 +542,7 @@ func mergeJSON(base, override json.RawMessage) (json.RawMessage, error) {
}
return json.Marshal(merged)
}
+
+// TypeCallMember is a member's state in a room's group call (MatrixRTC).
+// It is still under its MSC3401 name, which is what clients send.
+const TypeCallMember = "org.matrix.msc3401.call.member"
diff --git a/server/internal/setup/routes.go b/server/internal/setup/routes.go
index 829c9e3..25b5760 100644
--- a/server/internal/setup/routes.go
+++ b/server/internal/setup/routes.go
@@ -1,10 +1,12 @@
package setup
import (
+ "maps"
"net/http"
"time"
"github.com/dariuszjaniak/warp/server/internal/buildinfo"
+ "github.com/dariuszjaniak/warp/server/internal/clientapi"
"github.com/dariuszjaniak/warp/server/internal/httputil"
"github.com/dariuszjaniak/warp/server/internal/tracing"
)
@@ -32,10 +34,19 @@ type versionsResponse struct {
UnstableFeatures map[string]bool `json:"unstable_features,omitempty"`
}
+// unstableDelayedEvents is the MSC4140 feature flag. It is advertised only
+// while delayed events are switched on, because a MatrixRTC client checks
+// it before scheduling its delayed leave.
+const unstableDelayedEvents = "org.matrix.msc4140"
+
// wellKnownClient is the /.well-known/matrix/client document.
type wellKnownClient struct {
Homeserver wellKnownBaseURL `json:"m.homeserver"`
IdentityServer *wellKnownBaseURL `json:"m.identity_server,omitempty"`
+
+ // RTCFoci are the MatrixRTC backends (MSC4143) a client may use for a
+ // group call it starts.
+ RTCFoci []clientapi.RTCTransport `json:"org.matrix.msc4143.rtc_foci,omitempty"`
}
type wellKnownBaseURL struct {
@@ -84,10 +95,14 @@ func (s *Server) routes() *httputil.Router {
// handleVersions implements GET /_matrix/client/versions.
func (s *Server) handleVersions(w http.ResponseWriter, _ *http.Request) error {
+ features := maps.Clone(UnstableFeatures)
+ if s.delayed.Enabled() {
+ features[unstableDelayedEvents] = true
+ }
httputil.CacheFor(w, time.Hour)
httputil.WriteJSON(w, http.StatusOK, versionsResponse{
Versions: SpecVersions,
- UnstableFeatures: UnstableFeatures,
+ UnstableFeatures: features,
})
return nil
}
@@ -101,6 +116,7 @@ func (s *Server) handleWellKnownClient(w http.ResponseWriter, _ *http.Request) e
if url := s.cfg.Global.WellKnown.IdentityServerBaseURL; url != "" {
doc.IdentityServer = &wellKnownBaseURL{BaseURL: url}
}
+ doc.RTCFoci = clientapi.RTCTransports(s.cfg.ClientAPI.RTC.LiveKitServiceURL)
httputil.CacheFor(w, time.Hour)
httputil.WriteJSON(w, http.StatusOK, doc)
return nil
diff --git a/server/internal/setup/rtc_test.go b/server/internal/setup/rtc_test.go
new file mode 100644
index 0000000..40748ba
--- /dev/null
+++ b/server/internal/setup/rtc_test.go
@@ -0,0 +1,87 @@
+package setup_test
+
+import (
+ "net/http"
+ "testing"
+
+ "github.com/dariuszjaniak/warp/server/internal/config"
+)
+
+// TestWellKnownClientAdvertisesLiveKit: group call clients find the SFU
+// they publish to in the discovery document (MSC4143).
+func TestWellKnownClientAdvertisesLiveKit(t *testing.T) {
+ srv := newTestServer(t, func(c *config.Config) {
+ c.ClientAPI.RTC.LiveKitServiceURL = "https://livekit-jwt.warp.example/"
+ })
+ rec := get(t, srv, http.MethodGet, "/.well-known/matrix/client")
+ if rec.Code != http.StatusOK {
+ t.Fatalf("status = %d", rec.Code)
+ }
+ body := decode[struct {
+ Foci []map[string]string `json:"org.matrix.msc4143.rtc_foci"`
+ }](t, rec)
+ if len(body.Foci) != 1 || body.Foci[0]["type"] != "livekit" ||
+ body.Foci[0]["livekit_service_url"] != "https://livekit-jwt.warp.example" {
+ t.Errorf("rtc_foci = %v", body.Foci)
+ }
+}
+
+func TestWellKnownClientOmitsRTCFociByDefault(t *testing.T) {
+ srv := newTestServer(t, nil)
+ body := decode[map[string]any](t, get(t, srv, http.MethodGet, "/.well-known/matrix/client"))
+ if _, present := body["org.matrix.msc4143.rtc_foci"]; present {
+ t.Error("rtc_foci must be omitted when no LiveKit service is configured")
+ }
+}
+
+func TestRTCConfigValidation(t *testing.T) {
+ cfg := config.Default()
+ cfg.Global.ServerName = "warp.example"
+ cfg.ClientAPI.RTC.LiveKitServiceURL = "livekit-jwt.warp.example"
+ if err := cfg.Validate(); err == nil {
+ t.Error("a LiveKit service URL without a scheme was accepted")
+ }
+ cfg.ClientAPI.RTC.LiveKitServiceURL = ""
+ cfg.ClientAPI.DelayedEvents.MaxPerUser = 0
+ if err := cfg.Validate(); err == nil {
+ t.Error("delayed events with no room for any were accepted")
+ }
+}
+
+// TestGroupCallRoutesAreWiredUp: with a signing key (and so rooms) the
+// server advertises delayed events, serves their endpoints, and hands out
+// OpenID tokens the userinfo endpoint accepts.
+func TestGroupCallRoutesAreWiredUp(t *testing.T) {
+ srv := newSigningServer(t, func(c *config.Config) {
+ c.Registration.Enabled = true
+ c.ClientAPI.RateLimit.Enabled = false
+ })
+ features := decode[map[string]any](t, get(t, srv, http.MethodGet, "/_matrix/client/versions"))["unstable_features"].(map[string]any)
+ if features["org.matrix.msc4140"] != true {
+ t.Errorf("unstable_features = %v", features)
+ }
+
+ token := registerThrough(t, srv, "alice")
+ if rec := getWithToken(t, srv, "/_matrix/client/unstable/org.matrix.msc4140/delayed_events", token); rec.Code != http.StatusOK {
+ t.Errorf("delayed_events: %d %s", rec.Code, rec.Body.String())
+ }
+ rec := post(t, srv, "/_matrix/client/v3/user/@alice:warp.example/openid/request_token", map[string]any{}, token)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("request_token: %d %s", rec.Code, rec.Body.String())
+ }
+ openID := decode[map[string]any](t, rec)["access_token"].(string)
+ rec = get(t, srv, http.MethodGet, "/_matrix/federation/v1/openid/userinfo?access_token="+openID)
+ if rec.Code != http.StatusOK {
+ t.Errorf("userinfo: %d %s", rec.Code, rec.Body.String())
+ }
+}
+
+func TestDelayedEventsCanBeSwitchedOff(t *testing.T) {
+ srv := newSigningServer(t, func(c *config.Config) {
+ c.ClientAPI.DelayedEvents.MaxDelay = 0
+ })
+ features := decode[map[string]any](t, get(t, srv, http.MethodGet, "/_matrix/client/versions"))["unstable_features"].(map[string]any)
+ if _, present := features["org.matrix.msc4140"]; present {
+ t.Errorf("unstable_features = %v", features)
+ }
+}
diff --git a/server/internal/setup/server.go b/server/internal/setup/server.go
index 4ebc929..f39c31b 100644
--- a/server/internal/setup/server.go
+++ b/server/internal/setup/server.go
@@ -19,6 +19,7 @@ import (
"github.com/dariuszjaniak/warp/server/internal/buildinfo"
"github.com/dariuszjaniak/warp/server/internal/clientapi"
"github.com/dariuszjaniak/warp/server/internal/config"
+ "github.com/dariuszjaniak/warp/server/internal/delayedevents"
"github.com/dariuszjaniak/warp/server/internal/federation"
"github.com/dariuszjaniak/warp/server/internal/federationapi"
"github.com/dariuszjaniak/warp/server/internal/httputil"
@@ -61,6 +62,7 @@ type Server struct {
clientAPI *clientapi.ClientAPI
appservices *appservice.API
+ delayed *delayedevents.Service
notices *servernotices.API
regTokens *userapi.RegistrationTokens
@@ -407,6 +409,24 @@ func New(ctx context.Context, cfg *config.Config, opts Options) (*Server, error)
}
emit.add(s.appservices)
+ // Delayed events are sent by the roomserver as the user who
+ // scheduled them, and hear about state changes on the emit stage:
+ // someone else setting a piece of state cancels the delayed
+ // events that would overwrite it.
+ s.delayed, err = delayedevents.New(delayedevents.Options{
+ Store: s.db.DelayedEvents(),
+ Rooms: s.rooms,
+ MaxDelay: cfg.ClientAPI.DelayedEvents.MaxDelay,
+ MaxPerUser: cfg.ClientAPI.DelayedEvents.MaxPerUser,
+ DescribeError: clientapi.DescribeRoomError,
+ Logger: log,
+ })
+ if err != nil {
+ _ = s.closeDB()
+ return nil, err
+ }
+ emit.add(s.delayed)
+
// Deactivating an account leaves its rooms, which only the
// roomserver can do.
s.users.SetAccountObserver(roomLeaver{rooms: s.rooms, log: log})
@@ -490,6 +510,9 @@ func New(ctx context.Context, cfg *config.Config, opts Options) (*Server, error)
AppServices: s.appservices,
RateLimit: cfg.ClientAPI.RateLimit,
TURN: cfg.ClientAPI.TURN,
+ RTC: cfg.ClientAPI.RTC,
+ DelayedEvents: s.delayed,
+ OpenIDTokens: s.db.OpenIDTokens(),
TrustXForwardedFor: cfg.ClientAPI.TrustXForwardedFor,
Logger: log,
})
@@ -522,6 +545,11 @@ func New(ctx context.Context, cfg *config.Config, opts Options) (*Server, error)
// talks to the services, and tests that drive the handler directly
// need it running.
s.appservices.Start(ctx)
+ // So do delayed events: a call member who drops off has to be removed
+ // whether or not anything is listening yet.
+ if s.delayed != nil {
+ s.delayed.Start(ctx)
+ }
s.handler = httputil.Chain(s.routes(),
httputil.Recoverer,
@@ -749,6 +777,10 @@ func (s *Server) Close() error {
// cursors.
s.appservices.Close()
}
+ if s.delayed != nil {
+ // And the delayed event scheduler, which sends events.
+ s.delayed.Close()
+ }
if s.stopTracing != nil {
// Spans still in the batch are sent rather than lost.
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
diff --git a/server/internal/storage/delayed.go b/server/internal/storage/delayed.go
new file mode 100644
index 0000000..26a2410
--- /dev/null
+++ b/server/internal/storage/delayed.go
@@ -0,0 +1,110 @@
+package storage
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+)
+
+// ErrNoSuchDelayedEvent is returned for a delay ID that is unknown, belongs
+// to someone else, or is no longer scheduled.
+var ErrNoSuchDelayedEvent = errors.New("no such delayed event")
+
+// DelayedEvent is an event a client asked to have sent later (MSC4140).
+//
+// While it is scheduled FinalisedAtMS is zero. Once it has been sent or
+// cancelled it is kept for a while with its outcome, so a client that comes
+// back can find out what became of it.
+type DelayedEvent struct {
+ DelayID string
+ UserID string
+ DeviceID string
+ RoomID string
+ Type string
+
+ // StateKey is nil for a timeline event.
+ StateKey *string
+ Content json.RawMessage
+
+ // DelayMS is the delay the client asked for. A restart counts it again
+ // from now.
+ DelayMS int64
+
+ // RunningSinceMS is when the delay was last (re)started, and SendAtMS
+ // when it runs out: RunningSinceMS + DelayMS.
+ RunningSinceMS int64
+ SendAtMS int64
+
+ // The outcome, once finalised: Outcome is "send" or "cancel", Reason
+ // "delay", "action" or "error". EventID is the event that was sent,
+ // and Error the Matrix error that stopped it.
+ FinalisedAtMS int64
+ Outcome string
+ Reason string
+ EventID string
+ Error json.RawMessage
+}
+
+// DelayedEventStore keeps the delayed events of every user.
+type DelayedEventStore interface {
+ // InsertDelayedEvent schedules an event.
+ InsertDelayedEvent(ctx context.Context, e DelayedEvent) error
+
+ // DelayedEvent returns one of userID's events, scheduled or finalised.
+ DelayedEvent(ctx context.Context, userID, delayID string) (DelayedEvent, error)
+
+ // ScheduledDelayedEvent returns a scheduled event whoever owns it: the
+ // action endpoints are addressed by the delay ID alone.
+ ScheduledDelayedEvent(ctx context.Context, delayID string) (DelayedEvent, error)
+
+ // DelayedEvents lists userID's events, scheduled ones first by when
+ // they are due, then finalised ones newest first.
+ DelayedEvents(ctx context.Context, userID string) ([]DelayedEvent, error)
+
+ // CountScheduledDelayedEvents counts userID's scheduled events.
+ CountScheduledDelayedEvents(ctx context.Context, userID string) (int, error)
+
+ // RestartDelayedEvent starts a scheduled event's delay again from nowMS.
+ // It returns ErrNoSuchDelayedEvent if the event is not scheduled.
+ RestartDelayedEvent(ctx context.Context, delayID string, nowMS int64) error
+
+ // FinaliseDelayedEvent records what became of a scheduled event. It
+ // returns ErrNoSuchDelayedEvent if the event was already finalised,
+ // which is how two racing finalisers are told apart: only one of them
+ // gets to send.
+ FinaliseDelayedEvent(ctx context.Context, e DelayedEvent) error
+
+ // DueDelayedEvents returns up to limit scheduled events whose delay
+ // has run out at nowMS, the earliest first.
+ DueDelayedEvents(ctx context.Context, nowMS int64, limit int) ([]DelayedEvent, error)
+
+ // NextDelayedEventDue reports when the next scheduled event is due,
+ // and false when nothing is scheduled.
+ NextDelayedEventDue(ctx context.Context) (int64, bool, error)
+
+ // ScheduledDelayedStateEvents lists the scheduled state events for one
+ // piece of room state, of every user.
+ ScheduledDelayedStateEvents(ctx context.Context, roomID, eventType, stateKey string) ([]DelayedEvent, error)
+
+ // PurgeFinalisedDelayedEvents forgets the events finalised before
+ // beforeMS.
+ PurgeFinalisedDelayedEvents(ctx context.Context, beforeMS int64) error
+}
+
+// ErrNoSuchOpenIDToken is returned for an OpenID token that is unknown or
+// has expired.
+var ErrNoSuchOpenIDToken = errors.New("no such OpenID token")
+
+// OpenIDTokenStore keeps the OpenID tokens users hand to third parties, so
+// those can find out who the user is (and nothing more) from this server.
+type OpenIDTokenStore interface {
+ // InsertOpenIDToken stores a token by its hash.
+ InsertOpenIDToken(ctx context.Context, tokenHash, userID string, expiresAtMS int64) error
+
+ // OpenIDTokenUser returns the user a token was issued to, or
+ // ErrNoSuchOpenIDToken once it has expired at nowMS.
+ OpenIDTokenUser(ctx context.Context, tokenHash string, nowMS int64) (string, error)
+
+ // PurgeOpenIDTokens forgets tokens that expired before beforeMS.
+ PurgeOpenIDTokens(ctx context.Context, beforeMS int64) error
+}
diff --git a/server/internal/storage/delayed_test.go b/server/internal/storage/delayed_test.go
new file mode 100644
index 0000000..e121228
--- /dev/null
+++ b/server/internal/storage/delayed_test.go
@@ -0,0 +1,147 @@
+package storage_test
+
+import (
+ "database/sql"
+ "errors"
+ "os"
+ "testing"
+
+ "github.com/dariuszjaniak/warp/server/internal/storage"
+)
+
+// migratedDelayedDB opens a migrated database with empty delayed event and
+// OpenID tables.
+func migratedDelayedDB(t *testing.T, driver string) storage.Database {
+ t.Helper()
+ db := openTestDB(t, driver)
+ if err := db.Migrate(t.Context()); err != nil {
+ t.Fatalf("Migrate: %v", err)
+ }
+ if driver == storage.DriverPostgres {
+ pool, err := sql.Open("pgx", os.Getenv("WARP_TEST_POSTGRES_DSN"))
+ if err != nil {
+ t.Fatalf("open postgres: %v", err)
+ }
+ defer func() { _ = pool.Close() }()
+ if _, err := pool.ExecContext(t.Context(), `TRUNCATE delayed_events, openid_tokens`); err != nil {
+ t.Fatalf("reset the tables: %v", err)
+ }
+ }
+ return db
+}
+
+func TestDelayedEvents(t *testing.T) {
+ for _, driver := range testDrivers {
+ t.Run(driver, func(t *testing.T) {
+ ctx := t.Context()
+ store := migratedDelayedDB(t, driver).DelayedEvents()
+
+ if _, ok, err := store.NextDelayedEventDue(ctx); err != nil || ok {
+ t.Fatalf("NextDelayedEventDue on an empty table = %v, %v", ok, err)
+ }
+
+ key := "_@alice:warp.example_DEV_m.call"
+ state := storage.DelayedEvent{
+ DelayID: "d1", UserID: "@alice:warp.example", DeviceID: "DEV",
+ RoomID: "!r:warp.example", Type: "org.matrix.msc3401.call.member",
+ StateKey: &key, Content: []byte(`{}`),
+ DelayMS: 5000, RunningSinceMS: 1000, SendAtMS: 6000,
+ }
+ timeline := storage.DelayedEvent{
+ DelayID: "d2", UserID: "@alice:warp.example", RoomID: "!r:warp.example",
+ Type: "m.room.message", Content: []byte(`{"body":"hi"}`),
+ DelayMS: 1000, RunningSinceMS: 1000, SendAtMS: 2000,
+ }
+ for _, e := range []storage.DelayedEvent{state, timeline} {
+ if err := store.InsertDelayedEvent(ctx, e); err != nil {
+ t.Fatalf("InsertDelayedEvent: %v", err)
+ }
+ }
+
+ if n, err := store.CountScheduledDelayedEvents(ctx, "@alice:warp.example"); err != nil || n != 2 {
+ t.Errorf("count = %d, %v", n, err)
+ }
+ if next, ok, err := store.NextDelayedEventDue(ctx); err != nil || !ok || next != 2000 {
+ t.Errorf("next due = %d, %v, %v", next, ok, err)
+ }
+
+ got, err := store.DelayedEvent(ctx, "@alice:warp.example", "d1")
+ if err != nil || got.StateKey == nil || *got.StateKey != key || string(got.Content) != `{}` {
+ t.Errorf("DelayedEvent = %+v, %v", got, err)
+ }
+ if got, err := store.DelayedEvent(ctx, "@alice:warp.example", "d2"); err != nil || got.StateKey != nil {
+ t.Errorf("a timeline event has a state key: %+v, %v", got, err)
+ }
+ if _, err := store.DelayedEvent(ctx, "@bob:warp.example", "d1"); !errors.Is(err, storage.ErrNoSuchDelayedEvent) {
+ t.Errorf("bob read alice's event: %v", err)
+ }
+
+ due, err := store.DueDelayedEvents(ctx, 3000, 10)
+ if err != nil || len(due) != 1 || due[0].DelayID != "d2" {
+ t.Errorf("due at 3000 = %+v, %v", due, err)
+ }
+
+ if err := store.RestartDelayedEvent(ctx, "d1", 4000); err != nil {
+ t.Fatalf("RestartDelayedEvent: %v", err)
+ }
+ got, _ = store.ScheduledDelayedEvent(ctx, "d1")
+ if got.RunningSinceMS != 4000 || got.SendAtMS != 9000 {
+ t.Errorf("after restart = %+v", got)
+ }
+
+ pending, err := store.ScheduledDelayedStateEvents(ctx, "!r:warp.example", "org.matrix.msc3401.call.member", key)
+ if err != nil || len(pending) != 1 {
+ t.Errorf("ScheduledDelayedStateEvents = %+v, %v", pending, err)
+ }
+
+ sent := timeline
+ sent.FinalisedAtMS, sent.Outcome, sent.Reason, sent.EventID = 2500, "send", "delay", "$e"
+ if err := store.FinaliseDelayedEvent(ctx, sent); err != nil {
+ t.Fatalf("FinaliseDelayedEvent: %v", err)
+ }
+ if err := store.FinaliseDelayedEvent(ctx, sent); !errors.Is(err, storage.ErrNoSuchDelayedEvent) {
+ t.Errorf("finalising twice = %v", err)
+ }
+ if err := store.RestartDelayedEvent(ctx, "d2", 3000); !errors.Is(err, storage.ErrNoSuchDelayedEvent) {
+ t.Errorf("restarting a finalised event = %v", err)
+ }
+
+ all, err := store.DelayedEvents(ctx, "@alice:warp.example")
+ if err != nil || len(all) != 2 || all[0].DelayID != "d1" || all[1].EventID != "$e" {
+ t.Errorf("DelayedEvents = %+v, %v", all, err)
+ }
+
+ if err := store.PurgeFinalisedDelayedEvents(ctx, 3000); err != nil {
+ t.Fatal(err)
+ }
+ if all, _ := store.DelayedEvents(ctx, "@alice:warp.example"); len(all) != 1 {
+ t.Errorf("after purge = %+v", all)
+ }
+ })
+ }
+}
+
+func TestOpenIDTokens(t *testing.T) {
+ for _, driver := range testDrivers {
+ t.Run(driver, func(t *testing.T) {
+ ctx := t.Context()
+ store := migratedDelayedDB(t, driver).OpenIDTokens()
+
+ if err := store.InsertOpenIDToken(ctx, "hash", "@alice:warp.example", 5000); err != nil {
+ t.Fatal(err)
+ }
+ if user, err := store.OpenIDTokenUser(ctx, "hash", 4000); err != nil || user != "@alice:warp.example" {
+ t.Errorf("OpenIDTokenUser = %q, %v", user, err)
+ }
+ if _, err := store.OpenIDTokenUser(ctx, "hash", 6000); !errors.Is(err, storage.ErrNoSuchOpenIDToken) {
+ t.Errorf("an expired token = %v", err)
+ }
+ if err := store.PurgeOpenIDTokens(ctx, 6000); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := store.OpenIDTokenUser(ctx, "hash", 0); !errors.Is(err, storage.ErrNoSuchOpenIDToken) {
+ t.Errorf("a purged token = %v", err)
+ }
+ })
+ }
+}
diff --git a/server/internal/storage/postgres/migrations/0014_delayed_events_openid.sql b/server/internal/storage/postgres/migrations/0014_delayed_events_openid.sql
new file mode 100644
index 0000000..b009528
--- /dev/null
+++ b/server/internal/storage/postgres/migrations/0014_delayed_events_openid.sql
@@ -0,0 +1,50 @@
+-- Group calls: delayed events (MSC4140) and OpenID tokens.
+
+-- An event a client asked to have sent later. A MatrixRTC client schedules
+-- its own "left the call" this way and keeps pushing it back while it is
+-- in the call, so a client that crashes or loses its connection still
+-- leaves.
+--
+-- state_key is NULL for a timeline event. finalised_at_ms is 0 while the
+-- event is scheduled; once it has been sent or cancelled the row is kept
+-- for a while with its outcome, so a client can find out what happened.
+CREATE TABLE IF NOT EXISTS delayed_events (
+ delay_id TEXT NOT NULL PRIMARY KEY,
+ user_id TEXT NOT NULL,
+ device_id TEXT NOT NULL DEFAULT '',
+ room_id TEXT NOT NULL,
+ event_type TEXT NOT NULL,
+ state_key TEXT,
+ content TEXT NOT NULL,
+ delay_ms BIGINT NOT NULL,
+ running_since_ms BIGINT NOT NULL,
+ send_at_ms BIGINT NOT NULL,
+ finalised_at_ms BIGINT NOT NULL DEFAULT 0,
+ outcome TEXT NOT NULL DEFAULT '',
+ reason TEXT NOT NULL DEFAULT '',
+ event_id TEXT NOT NULL DEFAULT '',
+ error TEXT NOT NULL DEFAULT ''
+);
+
+CREATE INDEX IF NOT EXISTS delayed_events_user_idx
+ ON delayed_events (user_id, finalised_at_ms);
+
+-- The scheduler asks for what is due next among the scheduled events.
+CREATE INDEX IF NOT EXISTS delayed_events_due_idx
+ ON delayed_events (finalised_at_ms, send_at_ms);
+
+-- Someone else setting a piece of state cancels the delayed events that
+-- would have overwritten it.
+CREATE INDEX IF NOT EXISTS delayed_events_state_idx
+ ON delayed_events (room_id, event_type, state_key, finalised_at_ms);
+
+-- OpenID tokens: what a user hands a third party (the LiveKit JWT service)
+-- so it can ask this server who they are. Stored by hash, like access
+-- tokens.
+CREATE TABLE IF NOT EXISTS openid_tokens (
+ token_hash TEXT NOT NULL PRIMARY KEY,
+ user_id TEXT NOT NULL,
+ expires_at_ms BIGINT NOT NULL
+);
+
+CREATE INDEX IF NOT EXISTS openid_tokens_expiry_idx ON openid_tokens (expires_at_ms);
diff --git a/server/internal/storage/sqlite/migrations/0014_delayed_events_openid.sql b/server/internal/storage/sqlite/migrations/0014_delayed_events_openid.sql
new file mode 100644
index 0000000..b009528
--- /dev/null
+++ b/server/internal/storage/sqlite/migrations/0014_delayed_events_openid.sql
@@ -0,0 +1,50 @@
+-- Group calls: delayed events (MSC4140) and OpenID tokens.
+
+-- An event a client asked to have sent later. A MatrixRTC client schedules
+-- its own "left the call" this way and keeps pushing it back while it is
+-- in the call, so a client that crashes or loses its connection still
+-- leaves.
+--
+-- state_key is NULL for a timeline event. finalised_at_ms is 0 while the
+-- event is scheduled; once it has been sent or cancelled the row is kept
+-- for a while with its outcome, so a client can find out what happened.
+CREATE TABLE IF NOT EXISTS delayed_events (
+ delay_id TEXT NOT NULL PRIMARY KEY,
+ user_id TEXT NOT NULL,
+ device_id TEXT NOT NULL DEFAULT '',
+ room_id TEXT NOT NULL,
+ event_type TEXT NOT NULL,
+ state_key TEXT,
+ content TEXT NOT NULL,
+ delay_ms BIGINT NOT NULL,
+ running_since_ms BIGINT NOT NULL,
+ send_at_ms BIGINT NOT NULL,
+ finalised_at_ms BIGINT NOT NULL DEFAULT 0,
+ outcome TEXT NOT NULL DEFAULT '',
+ reason TEXT NOT NULL DEFAULT '',
+ event_id TEXT NOT NULL DEFAULT '',
+ error TEXT NOT NULL DEFAULT ''
+);
+
+CREATE INDEX IF NOT EXISTS delayed_events_user_idx
+ ON delayed_events (user_id, finalised_at_ms);
+
+-- The scheduler asks for what is due next among the scheduled events.
+CREATE INDEX IF NOT EXISTS delayed_events_due_idx
+ ON delayed_events (finalised_at_ms, send_at_ms);
+
+-- Someone else setting a piece of state cancels the delayed events that
+-- would have overwritten it.
+CREATE INDEX IF NOT EXISTS delayed_events_state_idx
+ ON delayed_events (room_id, event_type, state_key, finalised_at_ms);
+
+-- OpenID tokens: what a user hands a third party (the LiveKit JWT service)
+-- so it can ask this server who they are. Stored by hash, like access
+-- tokens.
+CREATE TABLE IF NOT EXISTS openid_tokens (
+ token_hash TEXT NOT NULL PRIMARY KEY,
+ user_id TEXT NOT NULL,
+ expires_at_ms BIGINT NOT NULL
+);
+
+CREATE INDEX IF NOT EXISTS openid_tokens_expiry_idx ON openid_tokens (expires_at_ms);
diff --git a/server/internal/storage/sqlstore/delayed.go b/server/internal/storage/sqlstore/delayed.go
new file mode 100644
index 0000000..5432c92
--- /dev/null
+++ b/server/internal/storage/sqlstore/delayed.go
@@ -0,0 +1,273 @@
+package sqlstore
+
+import (
+ "context"
+ "database/sql"
+ "errors"
+ "fmt"
+
+ "github.com/dariuszjaniak/warp/server/internal/storage"
+)
+
+// DelayedEvents is the shared implementation of storage.DelayedEventStore.
+type DelayedEvents struct {
+ db *sql.DB
+ rebind func(string) string
+}
+
+// NewDelayedEvents builds a delayed event store.
+func NewDelayedEvents(db *sql.DB, rebind func(string) string) *DelayedEvents {
+ if rebind == nil {
+ rebind = func(query string) string { return query }
+ }
+ return &DelayedEvents{db: db, rebind: rebind}
+}
+
+func (s *DelayedEvents) q(query string) string { return s.rebind(query) }
+
+const delayedEventColumns = `
+ delay_id, user_id, device_id, room_id, event_type, state_key, content,
+ delay_ms, running_since_ms, send_at_ms,
+ finalised_at_ms, outcome, reason, event_id, error`
+
+const (
+ insertDelayedEventSQL = `
+ INSERT INTO delayed_events (
+ delay_id, user_id, device_id, room_id, event_type, state_key, content,
+ delay_ms, running_since_ms, send_at_ms)
+ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`
+
+ selectDelayedEventSQL = `SELECT` + delayedEventColumns + `
+ FROM delayed_events WHERE user_id = ? AND delay_id = ?`
+
+ selectScheduledDelayedEventSQL = `SELECT` + delayedEventColumns + `
+ FROM delayed_events WHERE delay_id = ? AND finalised_at_ms = 0`
+
+ selectUserDelayedEventsSQL = `SELECT` + delayedEventColumns + `
+ FROM delayed_events WHERE user_id = ?
+ ORDER BY finalised_at_ms = 0 DESC, send_at_ms ASC, finalised_at_ms DESC`
+
+ countScheduledDelayedEventsSQL = `
+ SELECT COUNT(*) FROM delayed_events WHERE user_id = ? AND finalised_at_ms = 0`
+
+ restartDelayedEventSQL = `
+ UPDATE delayed_events SET running_since_ms = ?, send_at_ms = ? + delay_ms
+ WHERE delay_id = ? AND finalised_at_ms = 0`
+
+ finaliseDelayedEventSQL = `
+ UPDATE delayed_events
+ SET finalised_at_ms = ?, outcome = ?, reason = ?, event_id = ?, error = ?
+ WHERE delay_id = ? AND finalised_at_ms = 0`
+
+ selectDueDelayedEventsSQL = `SELECT` + delayedEventColumns + `
+ FROM delayed_events WHERE finalised_at_ms = 0 AND send_at_ms <= ?
+ ORDER BY send_at_ms LIMIT ?`
+
+ selectNextDelayedEventDueSQL = `
+ SELECT MIN(send_at_ms) FROM delayed_events WHERE finalised_at_ms = 0`
+
+ selectScheduledStateEventsSQL = `SELECT` + delayedEventColumns + `
+ FROM delayed_events
+ WHERE room_id = ? AND event_type = ? AND state_key = ? AND finalised_at_ms = 0`
+
+ purgeFinalisedDelayedEventsSQL = `
+ DELETE FROM delayed_events WHERE finalised_at_ms <> 0 AND finalised_at_ms < ?`
+)
+
+// InsertDelayedEvent implements storage.DelayedEventStore.
+func (s *DelayedEvents) InsertDelayedEvent(ctx context.Context, e storage.DelayedEvent) error {
+ var stateKey sql.NullString
+ if e.StateKey != nil {
+ stateKey = sql.NullString{String: *e.StateKey, Valid: true}
+ }
+ if _, err := s.db.ExecContext(ctx, s.q(insertDelayedEventSQL),
+ e.DelayID, e.UserID, e.DeviceID, e.RoomID, e.Type, stateKey, string(e.Content),
+ e.DelayMS, e.RunningSinceMS, e.SendAtMS,
+ ); err != nil {
+ return fmt.Errorf("insert delayed event: %w", err)
+ }
+ return nil
+}
+
+// DelayedEvent implements storage.DelayedEventStore.
+func (s *DelayedEvents) DelayedEvent(ctx context.Context, userID, delayID string) (storage.DelayedEvent, error) {
+ return s.one(ctx, selectDelayedEventSQL, userID, delayID)
+}
+
+// ScheduledDelayedEvent implements storage.DelayedEventStore.
+func (s *DelayedEvents) ScheduledDelayedEvent(ctx context.Context, delayID string) (storage.DelayedEvent, error) {
+ return s.one(ctx, selectScheduledDelayedEventSQL, delayID)
+}
+
+func (s *DelayedEvents) one(ctx context.Context, query string, args ...any) (storage.DelayedEvent, error) {
+ events, err := s.list(ctx, query, args...)
+ if err != nil {
+ return storage.DelayedEvent{}, err
+ }
+ if len(events) == 0 {
+ return storage.DelayedEvent{}, storage.ErrNoSuchDelayedEvent
+ }
+ return events[0], nil
+}
+
+// DelayedEvents implements storage.DelayedEventStore.
+func (s *DelayedEvents) DelayedEvents(ctx context.Context, userID string) ([]storage.DelayedEvent, error) {
+ return s.list(ctx, selectUserDelayedEventsSQL, userID)
+}
+
+// CountScheduledDelayedEvents implements storage.DelayedEventStore.
+func (s *DelayedEvents) CountScheduledDelayedEvents(ctx context.Context, userID string) (int, error) {
+ var n int
+ if err := s.db.QueryRowContext(ctx, s.q(countScheduledDelayedEventsSQL), userID).Scan(&n); err != nil {
+ return 0, fmt.Errorf("count delayed events: %w", err)
+ }
+ return n, nil
+}
+
+// RestartDelayedEvent implements storage.DelayedEventStore.
+func (s *DelayedEvents) RestartDelayedEvent(ctx context.Context, delayID string, nowMS int64) error {
+ res, err := s.db.ExecContext(ctx, s.q(restartDelayedEventSQL), nowMS, nowMS, delayID)
+ if err != nil {
+ return fmt.Errorf("restart delayed event: %w", err)
+ }
+ return oneRow(res)
+}
+
+// FinaliseDelayedEvent implements storage.DelayedEventStore.
+func (s *DelayedEvents) FinaliseDelayedEvent(ctx context.Context, e storage.DelayedEvent) error {
+ res, err := s.db.ExecContext(ctx, s.q(finaliseDelayedEventSQL),
+ e.FinalisedAtMS, e.Outcome, e.Reason, e.EventID, string(e.Error), e.DelayID)
+ if err != nil {
+ return fmt.Errorf("finalise delayed event: %w", err)
+ }
+ return oneRow(res)
+}
+
+func oneRow(res sql.Result) error {
+ n, err := res.RowsAffected()
+ if err != nil {
+ return fmt.Errorf("delayed event: %w", err)
+ }
+ if n == 0 {
+ return storage.ErrNoSuchDelayedEvent
+ }
+ return nil
+}
+
+// DueDelayedEvents implements storage.DelayedEventStore.
+func (s *DelayedEvents) DueDelayedEvents(ctx context.Context, nowMS int64, limit int) ([]storage.DelayedEvent, error) {
+ return s.list(ctx, selectDueDelayedEventsSQL, nowMS, limit)
+}
+
+// NextDelayedEventDue implements storage.DelayedEventStore.
+func (s *DelayedEvents) NextDelayedEventDue(ctx context.Context) (int64, bool, error) {
+ var next sql.NullInt64
+ if err := s.db.QueryRowContext(ctx, s.q(selectNextDelayedEventDueSQL)).Scan(&next); err != nil {
+ return 0, false, fmt.Errorf("next delayed event: %w", err)
+ }
+ return next.Int64, next.Valid, nil
+}
+
+// ScheduledDelayedStateEvents implements storage.DelayedEventStore.
+func (s *DelayedEvents) ScheduledDelayedStateEvents(ctx context.Context, roomID, eventType, stateKey string) ([]storage.DelayedEvent, error) {
+ return s.list(ctx, selectScheduledStateEventsSQL, roomID, eventType, stateKey)
+}
+
+// PurgeFinalisedDelayedEvents implements storage.DelayedEventStore.
+func (s *DelayedEvents) PurgeFinalisedDelayedEvents(ctx context.Context, beforeMS int64) error {
+ if _, err := s.db.ExecContext(ctx, s.q(purgeFinalisedDelayedEventsSQL), beforeMS); err != nil {
+ return fmt.Errorf("purge delayed events: %w", err)
+ }
+ return nil
+}
+
+func (s *DelayedEvents) list(ctx context.Context, query string, args ...any) ([]storage.DelayedEvent, error) {
+ rows, err := s.db.QueryContext(ctx, s.q(query), args...)
+ if err != nil {
+ return nil, fmt.Errorf("read delayed events: %w", err)
+ }
+ defer rows.Close()
+
+ var out []storage.DelayedEvent
+ for rows.Next() {
+ var (
+ e storage.DelayedEvent
+ stateKey sql.NullString
+ content, errorContent string
+ )
+ if err := rows.Scan(&e.DelayID, &e.UserID, &e.DeviceID, &e.RoomID, &e.Type, &stateKey, &content,
+ &e.DelayMS, &e.RunningSinceMS, &e.SendAtMS,
+ &e.FinalisedAtMS, &e.Outcome, &e.Reason, &e.EventID, &errorContent,
+ ); err != nil {
+ return nil, fmt.Errorf("read delayed event: %w", err)
+ }
+ if stateKey.Valid {
+ e.StateKey = &stateKey.String
+ }
+ e.Content = []byte(content)
+ if errorContent != "" {
+ e.Error = []byte(errorContent)
+ }
+ out = append(out, e)
+ }
+ return out, rows.Err()
+}
+
+// OpenIDTokens is the shared implementation of storage.OpenIDTokenStore.
+type OpenIDTokens struct {
+ db *sql.DB
+ rebind func(string) string
+}
+
+// NewOpenIDTokens builds an OpenID token store.
+func NewOpenIDTokens(db *sql.DB, rebind func(string) string) *OpenIDTokens {
+ if rebind == nil {
+ rebind = func(query string) string { return query }
+ }
+ return &OpenIDTokens{db: db, rebind: rebind}
+}
+
+const (
+ insertOpenIDTokenSQL = `
+ INSERT INTO openid_tokens (token_hash, user_id, expires_at_ms) VALUES (?, ?, ?)`
+
+ selectOpenIDTokenSQL = `
+ SELECT user_id FROM openid_tokens WHERE token_hash = ? AND expires_at_ms > ?`
+
+ purgeOpenIDTokensSQL = `DELETE FROM openid_tokens WHERE expires_at_ms < ?`
+)
+
+// InsertOpenIDToken implements storage.OpenIDTokenStore.
+func (s *OpenIDTokens) InsertOpenIDToken(ctx context.Context, tokenHash, userID string, expiresAtMS int64) error {
+ if _, err := s.db.ExecContext(ctx, s.rebind(insertOpenIDTokenSQL), tokenHash, userID, expiresAtMS); err != nil {
+ return fmt.Errorf("insert OpenID token: %w", err)
+ }
+ return nil
+}
+
+// OpenIDTokenUser implements storage.OpenIDTokenStore.
+func (s *OpenIDTokens) OpenIDTokenUser(ctx context.Context, tokenHash string, nowMS int64) (string, error) {
+ var userID string
+ err := s.db.QueryRowContext(ctx, s.rebind(selectOpenIDTokenSQL), tokenHash, nowMS).Scan(&userID)
+ if errors.Is(err, sql.ErrNoRows) {
+ return "", storage.ErrNoSuchOpenIDToken
+ }
+ if err != nil {
+ return "", fmt.Errorf("read OpenID token: %w", err)
+ }
+ return userID, nil
+}
+
+// PurgeOpenIDTokens implements storage.OpenIDTokenStore.
+func (s *OpenIDTokens) PurgeOpenIDTokens(ctx context.Context, beforeMS int64) error {
+ if _, err := s.db.ExecContext(ctx, s.rebind(purgeOpenIDTokensSQL), beforeMS); err != nil {
+ return fmt.Errorf("purge OpenID tokens: %w", err)
+ }
+ return nil
+}
+
+// Compile-time proof that the shared implementations satisfy the interfaces.
+var (
+ _ storage.DelayedEventStore = (*DelayedEvents)(nil)
+ _ storage.OpenIDTokenStore = (*OpenIDTokens)(nil)
+)
diff --git a/server/internal/storage/sqlstore/sqlstore.go b/server/internal/storage/sqlstore/sqlstore.go
index 290339f..84500c4 100644
--- a/server/internal/storage/sqlstore/sqlstore.go
+++ b/server/internal/storage/sqlstore/sqlstore.go
@@ -69,6 +69,8 @@ type Store struct {
push *Push
admin *Admin
apps *AppServices
+ delayed *DelayedEvents
+ openID *OpenIDTokens
}
// New builds a Store. It does not touch the database.
@@ -93,6 +95,8 @@ func New(opts Options) (*Store, error) {
push: NewPush(opts.DB, opts.Rebind),
admin: NewAdmin(opts.DB, opts.Rebind, opts.IsUniqueViolation),
apps: NewAppServices(opts.DB, opts.Rebind),
+ delayed: NewDelayedEvents(opts.DB, opts.Rebind),
+ openID: NewOpenIDTokens(opts.DB, opts.Rebind),
}, nil
}
@@ -138,6 +142,12 @@ func (s *Store) Admin() storage.AdminStore { return s.admin }
// AppServices implements storage.Database.
func (s *Store) AppServices() storage.AppServiceStore { return s.apps }
+// DelayedEvents implements storage.Database.
+func (s *Store) DelayedEvents() storage.DelayedEventStore { return s.delayed }
+
+// OpenIDTokens implements storage.Database.
+func (s *Store) OpenIDTokens() storage.OpenIDTokenStore { return s.openID }
+
// Migrate implements storage.Database.
func (s *Store) Migrate(ctx context.Context) error {
ms, err := s.migrations()
diff --git a/server/internal/storage/storage.go b/server/internal/storage/storage.go
index 400677e..34762f3 100644
--- a/server/internal/storage/storage.go
+++ b/server/internal/storage/storage.go
@@ -111,6 +111,14 @@ type Database interface {
// stream.
AppServices() AppServiceStore
+ // DelayedEvents is where the delayed events clients scheduled wait
+ // to be sent (MSC4140).
+ DelayedEvents() DelayedEventStore
+
+ // OpenIDTokens is where the OpenID tokens users hand to third parties
+ // are kept.
+ OpenIDTokens() OpenIDTokenStore
+
// Close releases the connection pool.
Close() error
}
diff --git a/warp.yaml.example b/warp.yaml.example
index 5770827..772a3c6 100644
--- a/warp.yaml.example
+++ b/warp.yaml.example
@@ -103,9 +103,11 @@ client_api:
# 0 means no expiry. WARP_REFRESH_TOKEN_LIFETIME
refresh_token_lifetime: 0s
- # Relay servers for voice and video calls, handed to clients by
- # GET /voip/turnServer. With no uris, calls only connect where the two
- # browsers can reach each other directly (the same network, mostly).
+ # Relay servers for Matrix 1:1 VoIP, handed to clients by
+ # GET /voip/turnServer. Warp's web client doesn't use them: its calls go
+ # through LiveKit (rtc below), which needs no TURN of its own here. This
+ # is for other Matrix clients that still place legacy 1:1 calls; with no
+ # uris, theirs only connect where the two can reach each other directly.
#
# With shared_secret (coturn's `use-auth-secret` + `static-auth-secret`)
# every user gets credentials that expire after user_lifetime. username
@@ -120,6 +122,24 @@ client_api:
username: "" # WARP_TURN_USERNAME
password: "" # WARP_TURN_PASSWORD
+ # Group calls (MatrixRTC). Warp carries no call media itself: clients
+ # publish to a LiveKit SFU and get a token for it from the LiveKit JWT
+ # service (lk-jwt-service), which checks who they are with an OpenID
+ # token from this server. Its URL is advertised to clients in
+ # /.well-known/matrix/client; leave it empty for no group calls.
+ # See docs/calls.md.
+ rtc:
+ livekit_service_url: "" # WARP_RTC_LIVEKIT_SERVICE_URL
+ # e.g. "https://livekit-jwt.example.org"
+
+ # Delayed events (MSC4140): events a client schedules now for the server
+ # to send later unless it pushes them back. A group call member's "left
+ # the call" is one, so someone whose browser crashes still leaves.
+ # max_delay 0 turns them off.
+ delayed_events:
+ max_delay: 24h # WARP_DELAYED_EVENTS_MAX_DELAY
+ max_per_user: 100 # WARP_DELAYED_EVENTS_MAX_PER_USER
+
# Per-endpoint rate limiting. Each bucket allows `burst` requests back
# to back and returns one slot every `interval`, per client address.
# Login and registration get their own, tighter, buckets because they