Compare commits

...
Author SHA1 Message Date
Jake Barnby 3e0df70f5c test(notifications): e2e tests for list, mark-read, tracking pixel, and webhook-failure fanout
Add 6 new test methods to NotificationsBase that exercise the wave3
account-alerts surface end-to-end:

- testListAccountAlertsEmpty: GET /v1/account/alerts shape check
- testWebhookFailureCreatesConsoleAlert: drives a webhook past
  _APP_WEBHOOK_MAX_FAILED_ATTEMPTS via user-create events and polls
  /account/alerts until the worker fans the paused alert out to the
  project owner on the console channel
- testMarkAlertReadTogglesFlag: PATCH /:alertId/read happy path
- testMarkAlertReadUnauthorized: stranger console user cannot mark
  someone else's alert as read; alert remains unread for the owner
- testTrackingPixelTogglesRead: GET /:alertId/track with a valid
  HS256 JWT signed with _APP_OPENSSL_KEY_V1 returns the canonical 1x1
  PNG and atomically marks the alert as read
- testTrackingPixelInvalidTokenReturnsPng: tampered JWT still gets a
  PNG (no information disclosure) but performs no DB write

Helpers:

- seedWebhookFailureAlert: registers a webhook pointing at an
  unroutable address (http://127.0.0.1:1/), drives max+2 user-create
  events through the project, polls assertEventually with a 60s budget
  for the paused alert keyed by the deterministic md5 of
  'webhook:<id>:paused:<attempts>'
- createConsoleUser: spins up a fresh, unrelated console user with its
  own session for the unauthorized assertion
- getConsoleAlertHeaders: console-session auth bundle reused across
  every alerts call, so the trait works identically under SideServer
  and SideConsole hosts
2026-05-06 15:09:28 +12:00
Jake Barnby 080b6dbeb2 merge: ST14 alerts Track body 2026-05-06 14:48:39 +12:00
Jake Barnby 678a6bd079 merge: ST13 alerts XList body 2026-05-06 14:48:36 +12:00
Jake Barnby 3b0c46c85e merge: ST12 PATCH /v1/account/alerts/:alertId/read 2026-05-06 14:48:33 +12:00
Jake Barnby c419257755 merge: ST11 Webhooks worker switch to Notification 2026-05-06 14:48:29 +12:00
Jake BarnbyandClaude Opus 4.7 067bf582dc feat(webhooks): emit pausing alerts via notifications worker to owners on email and console
When a webhook is permanently paused after exceeding the max failure
threshold, fan out the alert through the new Notifications worker
instead of the legacy Mails worker. Recipients are filtered to project
owners only and receive both an email and a console notification, with
a single deduplication key per webhook + attempts so duplicate triggers
collapse downstream.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-06 14:46:57 +12:00
Jake BarnbyandClaude Opus 4.7 dae98bf240 feat(notifications): track email opens by decoding pixel JWT and marking alert read
Replace placeholder body with JWT decode, alertId/userId match validation,
sparse update of read=true via injected Authorization::skip. Always returns
1x1 PNG regardless of decode outcome to avoid leaking JWT validity through
response status.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-06 14:46:33 +12:00
Jake BarnbyandClaude Opus 4.7 5dd987711b feat(notifications): implement GET /v1/account/alerts list endpoint body
Replace the placeholder body of the listAlerts action with real
implementation: scope alerts to the current user via userId equality,
parse incoming queries, support cursor pagination over the alerts
collection, run filters through count for total, and return
MODEL_ALERT_LIST.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-06 14:46:22 +12:00
Jake BarnbyandClaude Opus 4.7 e54fbf3b3f feat(notifications): mark alert read via PATCH /v1/account/alerts/:alertId/read
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-06 14:35:53 +12:00
Jake Barnby 0a9b54543d merge: ST10 queueForNotifications DI registration 2026-05-06 14:20:10 +12:00
Jake Barnby 93bdebb7fa merge: ST9 notifications worker fixes 2026-05-06 14:20:07 +12:00
Jake BarnbyandClaude Opus 4.7 6f01c1492f fix(notifications): worker dedup by attribute, throw on console-zero-delivery, thread recipient userId, inject email tracking pixel
Apply the four Greptile P1 fixes to the Notifications worker and
extend it for C3 email read tracking and ST4's stripped SMTP plumbing.

P1 #1: alreadyDelivered() now queries the indexed messageId attribute
instead of getDocument($messageId). The action loop and Console
adapter both write compound `$id`s (messageId + recipient hash), so
the previous direct-id lookup always missed.

P1 #3: action() no longer calls persistAlert after dispatchConsole;
ConsoleAdapter persists internally. Email persists inside
dispatchEmail BEFORE the adapter send so the alertId is available for
the tracking pixel; webhook persists in the action loop after a
successful HTTP send.

P1 #4: dispatchConsole now throws when the adapter reports
`deliveredTo === 0`, surfacing the per-recipient error.

Recipient threading: dispatch() now takes the full recipient map and
returns the alertId (or null when persistence is the caller's
responsibility). persistAlert() reads userId/teamId from the
recipient and grants per-user / per-team-owner CRUD permissions,
falling back to payload permissions only when neither is set. The
returned alertId lets dispatchEmail splice a 1x1 tracking pixel
before the last `</body>` tag, signed with a 30-day HS256 JWT
(_APP_OPENSSL_KEY_V1) carrying {alertId, userId}.

SMTP resolution: ST4 stripped `smtp` and `customMailOptions` from
the Notification event payload, so the worker now resolves SMTP
from the injected project Document (mirroring Mails.php /
Memberships/Create.php), falling back to the env-driven cloud SMTP
adapter when the project has no enabled override.

Tests updated: SpyNotifications.dispatch() matches the new
signature and emulates per-channel persistence so existing routing
assertions keep their semantics. Memory `alerts` collection adds
the `read` boolean attribute to mirror platform.php.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-06 14:16:25 +12:00
Jake BarnbyandClaude Opus 4.7 4e7236a366 feat(notifications): register queueForNotifications resource in worker and request containers
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-06 14:00:45 +12:00
Jake BarnbyandClaude Opus 4.7 8c7ecd6cdb fix(notifications): add SDK Method group to alerts skeleton actions
PHPStan flagged missing required `group` parameter on the SDK Method
constructors for the alerts XList and Track\Get skeleton actions.
Group set to 'alerts' to align with the resource grouping convention
used elsewhere in the account namespace.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-06 13:57:07 +12:00
Jake Barnby 0f1a2b5097 merge: ST4 event recipient + SMTP removal
# Conflicts:
#	src/Appwrite/Event/Notification.php
2026-05-06 13:48:47 +12:00
Jake Barnby a02eb3f41c merge: ST1 NOTIFICATION_TYPE_* rename 2026-05-06 13:48:16 +12:00
Jake Barnby ddffc69914 merge: ST8 alerts track skeleton
# Conflicts:
#	src/Appwrite/Platform/Modules/Account/Services/Http.php
2026-05-06 13:48:10 +12:00
Jake Barnby 439b3db2bf merge: ST7 alerts XList skeleton 2026-05-06 13:47:05 +12:00
Jake Barnby 134a8710da merge: ST6 alerts queries validator 2026-05-06 13:46:59 +12:00
Jake Barnby b9a928d3f8 merge: ST5 alert response model 2026-05-06 13:46:53 +12:00
Jake Barnby d07292ef1f merge: ST3 alerts schema 2026-05-06 13:46:49 +12:00
Jake Barnby 8cafc8c542 merge: ST2 console $id collision fix 2026-05-06 13:46:46 +12:00
Jake BarnbyandClaude Opus 4.7 a0400d2e93 feat(notifications): scaffold GET /v1/account/alerts/:alertId/track endpoint
Adds a public-scope tracking-pixel endpoint skeleton that always returns a
1x1 transparent PNG. The body is intentionally minimal so the email-tracking
read-flag flip can be wired up cleanly in Wave 3 (ST14) once the JWT decode
and DB write paths are designed.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-06 13:44:58 +12:00
Jake BarnbyandClaude Opus 4.7 d79be3a6e3 feat(notifications): add alert and alertList response models
Adds Alert response model and MODEL_ALERT/MODEL_ALERT_LIST constants
so future GET /v1/account/alerts endpoint can return MODEL_ALERT_LIST.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-06 13:44:27 +12:00
Jake BarnbyandClaude Opus 4.7 55275259d9 feat(notifications): scaffold GET /v1/account/alerts endpoint
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-06 13:44:09 +12:00
Jake BarnbyandClaude Opus 4.7 2c2e920a1a feat(notifications): add alerts queries validator
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-06 13:42:23 +12:00
Jake BarnbyandClaude Opus 4.7 4351dffb88 refactor(notifications): drop SMTP plumbing from event and extend recipient struct with userId/teamId
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-06 13:42:22 +12:00
Jake BarnbyandClaude Opus 4.7 a5587ff096 feat(notifications): track read state and enable per-recipient alert idempotency
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-06 13:42:10 +12:00
Jake BarnbyandClaude Opus 4.7 3df219df66 refactor(notifications): consolidate channel constants under NOTIFICATION_TYPE_* prefix
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-06 13:41:59 +12:00
Jake BarnbyandClaude Opus 4.7 36e1615498 fix(notifications): suffix Console adapter $id with recipient hash to avoid multi-recipient collision
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-06 13:41:11 +12:00
Jake BarnbyandClaude Opus 4.7 0ce3978f2c feat(notifications): add worker entrypoint and registration alongside Mails
Adds the bin/worker-notifications entrypoint, Dockerfile chmod, and
docker-compose service definitions for the new Notifications worker
without disturbing the existing Mails worker, and re-adds the Mails
queue/class constants on Event so the legacy callers still resolve.
The full mail->notification swap (caller migration, Mails removal,
worker rename) will land as a follow-up.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-01 15:56:43 +12:00
Jake Barnby d6be6e165b test(notifications): split e2e into base + server/console overlays
Mirrors the dominant e2e pattern (FunctionsBase/MigrationsBase): the
shared health-queue assertions live in a NotificationsBase trait,
with thin per-side overlays for ProjectCustom + SideServer and
ProjectCustom + SideConsole.
2026-05-01 15:51:25 +12:00
Jake Barnby 7839b3bb3f refactor(notifications): align webhook signing with per-recipient signatureKey
Drop the global _APP_NOTIFICATIONS_WEBHOOK_SECRET env var. There's no
analogous global webhook secret in Appwrite; the existing Webhooks
worker carries a per-webhook signatureKey on the webhook document.

Move the same pattern into the Notification event: each webhook
recipient may carry an optional signatureKey, which the worker
forwards to the Webhook adapter for HMAC-SHA256 signing. Recipients
without a key are delivered unsigned and a tag is logged for audit.
2026-05-01 15:51:25 +12:00
Jake Barnby 5320c9441b refactor(notifications): rename dedupKey to deduplicationKey
No abbreviations in identifier names.
2026-05-01 15:51:25 +12:00
Jake Barnby 8ad73aa6bb refactor(notifications): use Utopia Fetch client in Webhook adapter
Replace raw curl with Utopia\Fetch\Client to align with the rest of
the codebase (Github OAuth adapter, Install task). The dispatch seam
is preserved so existing tests can still substitute responses.
2026-05-01 15:51:25 +12:00
Jake Barnby 5cbbacaae5 refactor(notifications): move adapters under Appwrite\Utopia\Messaging
Mirror the upstream Utopia\Messaging package namespace under the
Appwrite\Utopia\Messaging prefix, matching the convention used by
Appwrite\Utopia\Database and Appwrite\Utopia\Response.
2026-05-01 15:51:25 +12:00
Jake Barnby 5ee92e1844 fix(notifications): move alerts collection to platform DB
Alerts is a platform-DB-only concern. Move from common.php to
platform.php where it belongs alongside other platform collections.
2026-05-01 15:51:25 +12:00
Jake BarnbyandClaude Opus 4.7 a546c41662 docs(notifications): document _APP_NOTIFICATIONS_WEBHOOK_SECRET
Registers the webhook signing secret in app/config/variables.php under a
new Notifications category, threads it through the worker container in
docker-compose.yml and tests/resources/docker/docker-compose.yml, and
adds an empty default to .env so operators see the knob alongside the
existing SMTP block.

When set, outbound webhook deliveries from the notifications worker
include an X-Appwrite-Webhook-Signature header carrying
sha256=<hex(hmac_sha256(timestamp.body, secret))>. When unset, the
worker delivers payloads unsigned — receivers must decide whether to
accept them.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-01 15:51:25 +12:00
Jake BarnbyandClaude Opus 4.7 df16c84b51 test(notifications): e2e coverage for notifications queue health
Adds the Notifications e2e suite under tests/e2e/Services/Notifications.
Asserts that the live notifications queue depth is reported via
GET /v1/health/queue/notifications, that the threshold guard is honoured,
and that the failed-jobs endpoint accepts the v1-notifications queue name.

Dispatch routing, dedup, and webhook signing are covered by the unit
suite — the worker cannot be deterministically driven through the live
queue from a test client without an admin enqueue endpoint, so the e2e
file pins the public health contract that ops dashboards and KEDA scale
on.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-01 15:51:24 +12:00
Jake BarnbyandClaude Opus 4.7 6bde54675e test(notifications): unit tests for worker and adapters
Covers the dedup short-circuit, per-channel dispatch routing, alert
persistence, error tagging, and legacy single-recipient fallback in the
Notifications worker, plus the Console adapter's permission shape and the
Webhook adapter's HMAC-SHA256 signing contract, header layout, response
handling, and unsigned-when-secret-missing behaviour.

Worker dispatch helpers move from private to protected so a test spy can
override them without monkey-patching. The Swoole runtime hook flag
mutation is now guarded by class_exists so the action can run under bare
PHPUnit (no Swoole extension on the test host).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-01 15:51:24 +12:00
Jake Barnby 90ef2d7487 feat(notifications): add Notifications worker and register it
Replaces the Mails worker. Resolves recipients into email, console, and
webhook channels, deduplicates deliveries via a hashed dedupKey, and persists
alerts for the console channel.
2026-05-01 12:07:42 +12:00
Jake Barnby 725ba86239 feat(notifications): add notifications schema and channel constants
Adds the alerts collection used by the console channel, plus channel and
provider type constants shared by the worker and adapters.
2026-05-01 12:07:40 +12:00
Jake Barnby 99641f24a5 feat(notifications): add Console and Webhook provider adapters
Console adapter persists alerts directly to the project database. Webhook
adapter dispatches signed JSON payloads (HMAC-SHA256) to subscriber URLs.
2026-05-01 12:07:38 +12:00
Jake Barnby 34935b8eed feat(notifications): add Notification event class
Replaces the Mail event with a multi-channel Notification event capable of
dispatching to email, console, and webhook recipients in a single payload.
Maintains the existing Mail surface (setSubject, setRecipient, setBody, etc.)
so existing call sites can be migrated mechanically.
2026-05-01 12:07:32 +12:00
32 changed files with 3070 additions and 21 deletions
+1
View File
@@ -93,6 +93,7 @@ RUN chmod +x /usr/local/bin/doctor && \
chmod +x /usr/local/bin/worker-functions && \
chmod +x /usr/local/bin/worker-mails && \
chmod +x /usr/local/bin/worker-messaging && \
chmod +x /usr/local/bin/worker-notifications && \
chmod +x /usr/local/bin/worker-migrations && \
chmod +x /usr/local/bin/worker-webhooks && \
chmod +x /usr/local/bin/worker-stats-usage && \
+144
View File
@@ -1011,6 +1011,150 @@ $platformCollections = [
],
],
'alerts' => [
'$collection' => ID::custom(Database::METADATA),
'$id' => ID::custom('alerts'),
'name' => 'Alerts',
'attributes' => [
[
'$id' => ID::custom('messageId'),
'type' => Database::VAR_STRING,
'format' => '',
'size' => Database::LENGTH_KEY,
'signed' => true,
'required' => false,
'default' => null,
'array' => false,
'filters' => [],
],
[
'$id' => ID::custom('type'),
'type' => Database::VAR_STRING,
'format' => '',
'size' => 64,
'signed' => true,
'required' => false,
'default' => 'info',
'array' => false,
'filters' => [],
],
[
'$id' => ID::custom('channel'),
'type' => Database::VAR_STRING,
'format' => '',
'size' => 64,
'signed' => true,
'required' => true,
'default' => null,
'array' => false,
'filters' => [],
],
[
'$id' => ID::custom('userId'),
'type' => Database::VAR_STRING,
'format' => '',
'size' => Database::LENGTH_KEY,
'signed' => true,
'required' => false,
'default' => null,
'array' => false,
'filters' => [],
],
[
'$id' => ID::custom('teamId'),
'type' => Database::VAR_STRING,
'format' => '',
'size' => Database::LENGTH_KEY,
'signed' => true,
'required' => false,
'default' => null,
'array' => false,
'filters' => [],
],
[
'$id' => ID::custom('projectId'),
'type' => Database::VAR_STRING,
'format' => '',
'size' => Database::LENGTH_KEY,
'signed' => true,
'required' => false,
'default' => null,
'array' => false,
'filters' => [],
],
[
'$id' => ID::custom('title'),
'type' => Database::VAR_STRING,
'format' => '',
'size' => 256,
'signed' => true,
'required' => true,
'default' => null,
'array' => false,
'filters' => [],
],
[
'$id' => ID::custom('body'),
'type' => Database::VAR_STRING,
'format' => '',
'size' => 16384,
'signed' => true,
'required' => true,
'default' => null,
'array' => false,
'filters' => [],
],
[
'$id' => ID::custom('read'),
'type' => Database::VAR_BOOLEAN,
'format' => '',
'size' => 0,
'signed' => true,
'required' => false,
'default' => false,
'array' => false,
'filters' => [],
],
],
'indexes' => [
[
'$id' => ID::custom('_key_messageId'),
'type' => Database::INDEX_KEY,
'attributes' => ['messageId'],
'lengths' => [Database::LENGTH_KEY],
'orders' => [Database::ORDER_ASC],
],
[
'$id' => ID::custom('_key_recipient'),
'type' => Database::INDEX_UNIQUE,
'attributes' => ['messageId', 'channel', 'userId', 'teamId'],
'lengths' => [Database::LENGTH_KEY, 64, Database::LENGTH_KEY, Database::LENGTH_KEY],
'orders' => [Database::ORDER_ASC, Database::ORDER_ASC, Database::ORDER_ASC, Database::ORDER_ASC],
],
[
'$id' => ID::custom('_key_userId_read'),
'type' => Database::INDEX_KEY,
'attributes' => ['userId', 'read'],
'lengths' => [Database::LENGTH_KEY, 0],
'orders' => [Database::ORDER_ASC, Database::ORDER_ASC],
],
[
'$id' => ID::custom('_key_teamId_read'),
'type' => Database::INDEX_KEY,
'attributes' => ['teamId', 'read'],
'lengths' => [Database::LENGTH_KEY, 0],
'orders' => [Database::ORDER_ASC, Database::ORDER_ASC],
],
[
'$id' => ID::custom('_key_projectId'),
'type' => Database::INDEX_KEY,
'attributes' => ['projectId'],
'lengths' => [Database::LENGTH_KEY],
'orders' => [Database::ORDER_ASC],
],
],
],
'certificates' => [
'$collection' => ID::custom(Database::METADATA),
'$id' => ID::custom('certificates'),
+7
View File
@@ -254,6 +254,13 @@ const FUNCTION_ALLOWLIST_HEADERS_RESPONSE = ['content-type', 'content-length'];
const MESSAGE_TYPE_EMAIL = 'email';
const MESSAGE_TYPE_SMS = 'sms';
const MESSAGE_TYPE_PUSH = 'push';
// Notification types
const NOTIFICATION_TYPE_EMAIL = 'email';
const NOTIFICATION_TYPE_SMS = 'sms';
const NOTIFICATION_TYPE_PUSH = 'push';
const NOTIFICATION_TYPE_CONSOLE = 'console';
const NOTIFICATION_TYPE_WEBHOOK = 'webhook';
const RESOURCE_TYPE_ALERTS = 'alerts';
// API key types
const API_KEY_STANDARD = 'standard';
const API_KEY_EPHEMERAL = 'ephemeral';
+3
View File
@@ -2,6 +2,7 @@
use Appwrite\Utopia\Response;
use Appwrite\Utopia\Response\Model\Account;
use Appwrite\Utopia\Response\Model\Alert;
use Appwrite\Utopia\Response\Model\AlgoArgon2;
use Appwrite\Utopia\Response\Model\AlgoBcrypt;
use Appwrite\Utopia\Response\Model\AlgoMd5;
@@ -237,6 +238,7 @@ Response::setModel(new BaseList('Column Indexes List', Response::MODEL_COLUMN_IN
Response::setModel(new BaseList('Users List', Response::MODEL_USER_LIST, 'users', Response::MODEL_USER));
Response::setModel(new BaseList('Sessions List', Response::MODEL_SESSION_LIST, 'sessions', Response::MODEL_SESSION));
Response::setModel(new BaseList('Identities List', Response::MODEL_IDENTITY_LIST, 'identities', Response::MODEL_IDENTITY));
Response::setModel(new BaseList('Alerts List', Response::MODEL_ALERT_LIST, 'alerts', Response::MODEL_ALERT));
Response::setModel(new BaseList('Logs List', Response::MODEL_LOG_LIST, 'logs', Response::MODEL_LOG));
Response::setModel(new BaseList('Files List', Response::MODEL_FILE_LIST, 'files', Response::MODEL_FILE));
Response::setModel(new BaseList('Buckets List', Response::MODEL_BUCKET_LIST, 'buckets', Response::MODEL_BUCKET));
@@ -362,6 +364,7 @@ Response::setModel(new Account());
Response::setModel(new Preferences());
Response::setModel(new Session());
Response::setModel(new Identity());
Response::setModel(new Alert());
Response::setModel(new Token());
Response::setModel(new JWT());
Response::setModel(new Locale());
+4
View File
@@ -12,6 +12,7 @@ use Appwrite\Event\Event;
use Appwrite\Event\Func;
use Appwrite\Event\Mail;
use Appwrite\Event\Messaging;
use Appwrite\Event\Notification;
use Appwrite\Event\Realtime;
use Appwrite\Event\Webhook;
use Appwrite\Extend\Exception;
@@ -123,6 +124,9 @@ return function (Container $container): void {
$container->set('queueForMails', function (Publisher $publisher) {
return new Mail($publisher);
}, ['publisher']);
$container->set('queueForNotifications', function (Publisher $publisher) {
return new Notification($publisher);
}, ['publisher']);
$container->set('queueForBuilds', function (Publisher $publisher) {
return new Build($publisher);
}, ['publisher']);
+5
View File
@@ -7,6 +7,7 @@ use Appwrite\Event\Event;
use Appwrite\Event\Func;
use Appwrite\Event\Mail;
use Appwrite\Event\Messaging;
use Appwrite\Event\Notification;
use Appwrite\Event\Realtime;
use Appwrite\Event\Webhook;
use Appwrite\Usage\Context;
@@ -342,6 +343,10 @@ return function (Container $container): void {
return new Mail($publisher);
}, ['publisher']);
$container->set('queueForNotifications', function (Publisher $publisher) {
return new Notification($publisher);
}, ['publisher']);
$container->set('queueForBuilds', function (Publisher $publisher) {
return new Build($publisher);
}, ['publisher']);
+3
View File
@@ -0,0 +1,3 @@
#!/bin/sh
exec php /usr/src/code/app/worker.php notifications "$@"
+41
View File
@@ -806,6 +806,47 @@ services:
- _APP_OPTIONS_FORCE_HTTPS
- _APP_DATABASE_SHARED_TABLES
appwrite-worker-notifications:
entrypoint: worker-notifications
<<: [*x-logging, *x-build]
container_name: appwrite-worker-notifications
image: appwrite-dev
networks:
- appwrite
volumes:
- ./app:/usr/src/code/app
- ./src:/usr/src/code/src
depends_on:
- redis
- maildev
- ${_APP_DB_HOST:-mariadb}
environment:
- _APP_ENV
- _APP_WORKER_PER_CORE
- _APP_POOL_ADAPTER
- _APP_OPENSSL_KEY_V1
- _APP_SYSTEM_EMAIL_NAME
- _APP_SYSTEM_EMAIL_ADDRESS
- _APP_REDIS_HOST
- _APP_REDIS_PORT
- _APP_REDIS_USER
- _APP_REDIS_PASS
- _APP_DB_HOST
- _APP_DB_PORT
- _APP_DB_SCHEMA
- _APP_DB_USER
- _APP_DB_PASS
- _APP_DB_ADAPTER
- _APP_SMTP_HOST
- _APP_SMTP_PORT
- _APP_SMTP_SECURE
- _APP_SMTP_USERNAME
- _APP_SMTP_PASSWORD
- _APP_LOGGING_CONFIG
- _APP_DOMAIN
- _APP_OPTIONS_FORCE_HTTPS
- _APP_DATABASE_SHARED_TABLES
appwrite-worker-messaging:
entrypoint: worker-messaging
<<: [*x-logging, *x-build]
+1
View File
@@ -37,6 +37,7 @@
<directory>./tests/e2e/Services/ProjectWebhooks</directory>
<directory>./tests/e2e/Services/Messaging</directory>
<directory>./tests/e2e/Services/Migrations</directory>
<directory>./tests/e2e/Services/Notifications</directory>
<directory>./tests/e2e/Services/Project</directory>
<file>./tests/e2e/Services/Functions/FunctionsBase.php</file>
<file>./tests/e2e/Services/Functions/FunctionsCustomServerTest.php</file>
+3
View File
@@ -21,6 +21,9 @@ class Event
public const MAILS_QUEUE_NAME = 'v1-mails';
public const MAILS_CLASS_NAME = 'MailsV1';
public const NOTIFICATIONS_QUEUE_NAME = 'v1-notifications';
public const NOTIFICATIONS_CLASS_NAME = 'NotificationsV1';
public const FUNCTIONS_QUEUE_NAME = 'v1-functions';
public const FUNCTIONS_CLASS_NAME = 'FunctionsV1';
public const FUNCTIONS_QUEUE_TTL = 60 * 60 * 24 * 7; // 7 days
+322
View File
@@ -0,0 +1,322 @@
<?php
namespace Appwrite\Event;
use Utopia\Config\Config;
use Utopia\Queue\Publisher;
use Utopia\System\System;
class Notification extends Event
{
protected string $recipient = '';
protected string $name = '';
protected string $subject = '';
protected string $body = '';
protected string $preview = '';
protected array $variables = [];
protected string $bodyTemplate = '';
protected array $attachment = [];
/**
* Recipients to deliver the notification to.
*
* Each entry has an `address` (channel-specific identifier — email,
* userId, or webhook URL) and a `channel`. Webhook recipients may
* additionally carry an optional `signatureKey`; when set, the
* webhook adapter signs the request body with HMAC-SHA256 and adds
* the `X-Appwrite-Webhook-Signature` header. Without a key the
* payload is sent unsigned. Optional `userId` and `teamId` identify
* the owner of the alert (used by C2/C3 budget/limit alerts).
*
* @var array<int, array{address: string, channel: string, signatureKey?: string, userId?: string, teamId?: string}>
*/
protected array $recipients = [];
/**
* @var array<string>
*/
protected array $channels = [];
protected string $template = '';
protected array $templateParams = [];
protected string $deduplicationKey = '';
/**
* @var array<string>
*/
protected array $permissions = [];
public function __construct(protected Publisher $publisher)
{
parent::__construct($publisher);
$this
->setQueue(System::getEnv('_APP_NOTIFICATIONS_QUEUE_NAME', Event::NOTIFICATIONS_QUEUE_NAME))
->setClass(System::getEnv('_APP_NOTIFICATIONS_CLASS_NAME', Event::NOTIFICATIONS_CLASS_NAME));
}
public function setSubject(string $subject): self
{
$this->subject = $subject;
return $this;
}
public function getSubject(): string
{
return $this->subject;
}
public function setRecipient(string $recipient): self
{
$this->recipient = $recipient;
return $this;
}
public function getRecipient(): string
{
return $this->recipient;
}
public function setBody(string $body): self
{
$this->body = $body;
return $this;
}
public function getBody(): string
{
return $this->body;
}
public function setPreview(string $preview): self
{
$this->preview = $preview;
return $this;
}
public function getPreview(): string
{
return $this->preview;
}
public function setName(string $name): self
{
$this->name = $name;
return $this;
}
public function getName(): string
{
return $this->name;
}
public function setBodyTemplate(string $bodyTemplate): self
{
$this->bodyTemplate = $bodyTemplate;
return $this;
}
public function getBodyTemplate(): string
{
return $this->bodyTemplate;
}
public function getVariables(): array
{
return $this->variables;
}
public function setVariables(array $variables): self
{
$this->variables = $variables;
return $this;
}
public function appendVariables(array $variables): self
{
$this->variables = \array_merge($this->variables, $variables);
return $this;
}
public function setAttachment(string $content, string $filename, string $encoding = 'base64', string $type = 'plain/text'): self
{
$this->attachment = [
'content' => \base64_encode($content),
'filename' => $filename,
'encoding' => $encoding,
'type' => $type,
];
return $this;
}
public function getAttachment(): array
{
return $this->attachment;
}
public function resetAttachment(): self
{
$this->attachment = [];
return $this;
}
/**
* @param array<int, array{address: string, channel: string, signatureKey?: string, userId?: string, teamId?: string}> $recipients
*/
public function setRecipients(array $recipients): self
{
$this->recipients = $recipients;
return $this;
}
/**
* @return array<int, array{address: string, channel: string, signatureKey?: string, userId?: string, teamId?: string}>
*/
public function getRecipients(): array
{
return $this->recipients;
}
public function addRecipient(
string $address,
string $channel = NOTIFICATION_TYPE_EMAIL,
?string $signatureKey = null,
?string $userId = null,
?string $teamId = null,
): self {
$recipient = ['address' => $address, 'channel' => $channel];
if ($signatureKey !== null && $signatureKey !== '') {
$recipient['signatureKey'] = $signatureKey;
}
if ($userId !== null && $userId !== '') {
$recipient['userId'] = $userId;
}
if ($teamId !== null && $teamId !== '') {
$recipient['teamId'] = $teamId;
}
$this->recipients[] = $recipient;
return $this;
}
/**
* @param array<string> $channels
*/
public function setChannels(array $channels): self
{
$this->channels = $channels;
return $this;
}
/**
* @return array<string>
*/
public function getChannels(): array
{
return $this->channels;
}
public function setTemplate(string $template): self
{
$this->template = $template;
return $this;
}
public function getTemplate(): string
{
return $this->template;
}
public function setTemplateParams(array $params): self
{
$this->templateParams = $params;
return $this;
}
public function getTemplateParams(): array
{
return $this->templateParams;
}
public function setDeduplicationKey(string $deduplicationKey): self
{
$this->deduplicationKey = $deduplicationKey;
return $this;
}
public function getDeduplicationKey(): string
{
return $this->deduplicationKey;
}
/**
* @param array<string> $permissions
*/
public function setPermissions(array $permissions): self
{
$this->permissions = $permissions;
return $this;
}
/**
* @return array<string>
*/
public function getPermissions(): array
{
return $this->permissions;
}
public function reset(): self
{
$this->project = null;
$this->recipient = '';
$this->name = '';
$this->subject = '';
$this->body = '';
$this->variables = [];
$this->bodyTemplate = '';
$this->attachment = [];
$this->recipients = [];
$this->channels = [];
$this->template = '';
$this->templateParams = [];
$this->deduplicationKey = '';
$this->permissions = [];
return $this;
}
protected function preparePayload(): array
{
$platform = $this->platform;
if (empty($platform)) {
$platform = Config::getParam('platform', []);
}
$recipients = $this->recipients;
if (empty($recipients) && !empty($this->recipient)) {
$recipients = [[
'address' => $this->recipient,
'channel' => NOTIFICATION_TYPE_EMAIL,
]];
}
return [
'project' => $this->project,
'recipient' => $this->recipient,
'recipients' => $recipients,
'channels' => $this->channels,
'template' => $this->template,
'templateParams' => $this->templateParams,
'deduplicationKey' => $this->deduplicationKey,
'permissions' => $this->permissions,
'name' => $this->name,
'subject' => $this->subject,
'bodyTemplate' => $this->bodyTemplate,
'body' => $this->body,
'preview' => $this->preview,
'variables' => $this->variables,
'attachment' => $this->attachment,
'events' => Event::generateEvents($this->getEvent(), $this->getParams()),
'platform' => $platform,
];
}
}
@@ -0,0 +1,75 @@
<?php
namespace Appwrite\Platform\Modules\Account\Http\Alerts\Read;
use Appwrite\Extend\Exception;
use Appwrite\SDK\AuthType;
use Appwrite\SDK\Method;
use Appwrite\SDK\Response as SDKResponse;
use Appwrite\Utopia\Response;
use Utopia\Database\Database;
use Utopia\Database\Document;
use Utopia\Database\Validator\UID;
use Utopia\Platform\Action;
use Utopia\Platform\Scope\HTTP;
class Update extends Action
{
use HTTP;
public static function getName(): string
{
return 'updateAlertRead';
}
public function __construct()
{
$this
->setHttpMethod(Action::HTTP_REQUEST_METHOD_PATCH)
->setHttpPath('/v1/account/alerts/:alertId/read')
->desc('Mark alert read')
->groups(['api', 'account'])
->label('scope', 'account')
->label('sdk', new Method(
namespace: 'account',
group: 'alerts',
name: 'updateAlertRead',
description: '/docs/references/account/update-alert-read.md',
auth: [AuthType::SESSION, AuthType::JWT],
responses: [
new SDKResponse(
code: Response::STATUS_CODE_OK,
model: Response::MODEL_ALERT,
)
]
))
->param('alertId', '', new UID(), 'Alert ID.')
->inject('response')
->inject('dbForPlatform')
->inject('user')
->callback($this->action(...));
}
public function action(
string $alertId,
Response $response,
Database $dbForPlatform,
Document $user,
): void {
$alert = $dbForPlatform->getDocument('alerts', $alertId);
if ($alert->isEmpty()) {
throw new Exception(Exception::DOCUMENT_NOT_FOUND);
}
if ($alert->getAttribute('userId') !== $user->getId()) {
throw new Exception(Exception::USER_UNAUTHORIZED);
}
$updated = $dbForPlatform->updateDocument('alerts', $alertId, new Document([
'read' => true,
]));
$response->dynamic($updated, Response::MODEL_ALERT);
}
}
@@ -0,0 +1,104 @@
<?php
namespace Appwrite\Platform\Modules\Account\Http\Alerts\Track;
use Ahc\Jwt\JWT;
use Appwrite\SDK\AuthType;
use Appwrite\SDK\ContentType;
use Appwrite\SDK\Method;
use Appwrite\SDK\Response as SDKResponse;
use Appwrite\Utopia\Response;
use Utopia\Database\Database;
use Utopia\Database\Document;
use Utopia\Database\Validator\Authorization;
use Utopia\Database\Validator\UID;
use Utopia\Platform\Action;
use Utopia\Platform\Scope\HTTP;
use Utopia\System\System;
use Utopia\Validator\Text;
class Get extends Action
{
use HTTP;
public static function getName(): string
{
return 'getAlertTrack';
}
public function __construct()
{
$this
->setHttpMethod(Action::HTTP_REQUEST_METHOD_GET)
->setHttpPath('/v1/account/alerts/:alertId/track')
->desc('Track alert')
->groups(['api', 'account'])
->label('scope', 'public')
->label('sdk', new Method(
namespace: 'account',
group: 'alerts',
name: 'getAlertTrack',
description: '/docs/references/account/get-alert-track.md',
auth: [AuthType::SESSION, AuthType::JWT],
responses: [
new SDKResponse(
code: Response::STATUS_CODE_OK,
model: Response::MODEL_NONE,
),
],
contentType: ContentType::IMAGE_PNG,
))
->param('alertId', '', new UID(), 'Alert ID.')
->param('jwt', '', new Text(2048, 0), 'Tracking token.', true)
->inject('response')
->inject('dbForPlatform')
->inject('authorization')
->callback($this->action(...));
}
public function action(
string $alertId,
string $jwt,
Response $response,
Database $dbForPlatform,
Authorization $authorization,
): void {
$secret = System::getEnv('_APP_OPENSSL_KEY_V1');
if ($secret !== '' && $jwt !== '') {
try {
$decoder = new JWT($secret, 'HS256', 2592000, 0);
$decoded = $decoder->decode($jwt);
if (
isset($decoded['alertId'], $decoded['userId'])
&& $decoded['alertId'] === $alertId
) {
$authorization->skip(function () use ($dbForPlatform, $alertId, $decoded) {
$alert = $dbForPlatform->getDocument('alerts', $alertId);
if (
!$alert->isEmpty()
&& $alert->getAttribute('userId') === $decoded['userId']
&& $alert->getAttribute('read') !== true
) {
$dbForPlatform->updateDocument('alerts', $alertId, new Document([
'read' => true,
]));
}
});
}
} catch (\Throwable) {
// Silent fail — never reveal JWT validity through response status
}
}
// 1x1 transparent PNG (canonical 67-byte payload)
$pixel = base64_decode('iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8/5+hHgAHggJ/PchI7wAAAABJRU5ErkJggg==');
$response
->setContentType('image/png')
->addHeader('Cache-Control', 'no-store')
->send($pixel);
}
}
@@ -0,0 +1,113 @@
<?php
namespace Appwrite\Platform\Modules\Account\Http\Alerts;
use Appwrite\Extend\Exception;
use Appwrite\SDK\AuthType;
use Appwrite\SDK\Method;
use Appwrite\SDK\Response as SDKResponse;
use Appwrite\Utopia\Database\Validator\Queries\Alerts;
use Appwrite\Utopia\Response;
use Utopia\Database\Database;
use Utopia\Database\Document;
use Utopia\Database\Exception\Order as OrderException;
use Utopia\Database\Exception\Query as QueryException;
use Utopia\Database\Query;
use Utopia\Database\Validator\Query\Cursor;
use Utopia\Platform\Action;
use Utopia\Platform\Scope\HTTP;
class XList extends Action
{
use HTTP;
public static function getName(): string
{
return 'listAlerts';
}
public function __construct()
{
$this
->setHttpMethod(Action::HTTP_REQUEST_METHOD_GET)
->setHttpPath('/v1/account/alerts')
->desc('List alerts')
->groups(['api', 'account'])
->label('scope', 'account')
->label('sdk', new Method(
namespace: 'account',
group: 'alerts',
name: 'listAlerts',
description: '/docs/references/account/list-alerts.md',
auth: [AuthType::SESSION, AuthType::JWT],
responses: [
new SDKResponse(
code: Response::STATUS_CODE_OK,
model: Response::MODEL_ALERT_LIST,
)
]
))
->param('queries', [], new Alerts(), 'Array of query strings generated using the Query class provided by the SDK. [Learn more about queries](https://appwrite.io/docs/queries). Maximum of ' . APP_LIMIT_ARRAY_PARAMS_SIZE . ' queries are allowed, each ' . APP_LIMIT_ARRAY_ELEMENT_SIZE . ' characters long. You may filter on the following attributes: ' . implode(', ', Alerts::ALLOWED_ATTRIBUTES), true)
->inject('response')
->inject('dbForPlatform')
->inject('user')
->callback($this->action(...));
}
/**
* @param array<string> $queries
*/
public function action(
array $queries,
Response $response,
Database $dbForPlatform,
Document $user
): void {
try {
$queries = Query::parseQueries($queries);
} catch (QueryException $e) {
throw new Exception(Exception::GENERAL_QUERY_INVALID, $e->getMessage());
}
$queries[] = Query::equal('userId', [$user->getId()]);
/**
* Get cursor document if there was a cursor query, we use array_filter and reset for reference $cursor to $queries
*/
$cursor = \array_filter($queries, function ($query) {
return \in_array($query->getMethod(), [Query::TYPE_CURSOR_AFTER, Query::TYPE_CURSOR_BEFORE]);
});
$cursor = reset($cursor);
if ($cursor) {
/** @var Query $cursor */
$validator = new Cursor();
if (!$validator->isValid($cursor)) {
throw new Exception(Exception::GENERAL_QUERY_INVALID, $validator->getDescription());
}
$alertId = $cursor->getValue();
$cursorDocument = $dbForPlatform->getDocument('alerts', $alertId);
if ($cursorDocument->isEmpty()) {
throw new Exception(Exception::GENERAL_CURSOR_NOT_FOUND, "Alert '{$alertId}' for the 'cursor' value not found.");
}
$cursor->setValue($cursorDocument);
}
$filterQueries = Query::groupByType($queries)['filters'];
try {
$results = $dbForPlatform->find('alerts', $queries);
$total = $dbForPlatform->count('alerts', $filterQueries, APP_LIMIT_COUNT);
} catch (OrderException $e) {
throw new Exception(Exception::DATABASE_QUERY_ORDER_NULL, "The order attribute '{$e->getAttribute()}' had a null value. Cursor pagination requires all documents order attribute values are non-null.");
}
$response->dynamic(new Document([
'alerts' => $results,
'total' => $total,
]), Response::MODEL_ALERT_LIST);
}
}
@@ -12,6 +12,9 @@ use Appwrite\Platform\Modules\Account\Http\Account\MFA\RecoveryCodes\Create as C
use Appwrite\Platform\Modules\Account\Http\Account\MFA\RecoveryCodes\Get as GetRecoveryCodes;
use Appwrite\Platform\Modules\Account\Http\Account\MFA\RecoveryCodes\Update as UpdateRecoveryCodes;
use Appwrite\Platform\Modules\Account\Http\Account\MFA\Update as UpdateMfa;
use Appwrite\Platform\Modules\Account\Http\Alerts\Read\Update as UpdateAlertRead;
use Appwrite\Platform\Modules\Account\Http\Alerts\Track\Get as TrackAlert;
use Appwrite\Platform\Modules\Account\Http\Alerts\XList as ListAlerts;
use Utopia\Platform\Service;
class Http extends Service
@@ -29,6 +32,9 @@ class Http extends Service
->addAction(UpdateRecoveryCodes::getName(), new UpdateRecoveryCodes())
->addAction(GetRecoveryCodes::getName(), new GetRecoveryCodes())
->addAction(CreateChallenge::getName(), new CreateChallenge())
->addAction(UpdateChallenge::getName(), new UpdateChallenge());
->addAction(UpdateChallenge::getName(), new UpdateChallenge())
->addAction(ListAlerts::getName(), new ListAlerts())
->addAction(UpdateAlertRead::getName(), new UpdateAlertRead())
->addAction(TrackAlert::getName(), new TrackAlert());
}
}
@@ -10,6 +10,7 @@ use Appwrite\Platform\Workers\Functions;
use Appwrite\Platform\Workers\Mails;
use Appwrite\Platform\Workers\Messaging;
use Appwrite\Platform\Workers\Migrations;
use Appwrite\Platform\Workers\Notifications;
use Appwrite\Platform\Workers\StatsResources;
use Appwrite\Platform\Workers\StatsUsage;
use Appwrite\Platform\Workers\Webhooks;
@@ -28,6 +29,7 @@ class Workers extends Service
->addAction(Functions::getName(), new Functions())
->addAction(Mails::getName(), new Mails())
->addAction(Messaging::getName(), new Messaging())
->addAction(Notifications::getName(), new Notifications())
->addAction(Webhooks::getName(), new Webhooks())
->addAction(StatsUsage::getName(), new StatsUsage())
->addAction(Migrations::getName(), new Migrations())
@@ -0,0 +1,560 @@
<?php
namespace Appwrite\Platform\Workers;
use Ahc\Jwt\JWT;
use Appwrite\Template\Template;
use Appwrite\Utopia\Messaging\Adapter\Console as ConsoleAdapter;
use Appwrite\Utopia\Messaging\Adapter\Webhook as WebhookAdapter;
use Appwrite\Utopia\Messaging\Messages\Console as ConsoleMessage;
use Appwrite\Utopia\Messaging\Messages\Webhook as WebhookMessage;
use Exception;
use Swoole\Runtime;
use Throwable;
use Utopia\Database\Database;
use Utopia\Database\Document;
use Utopia\Database\Exception\Duplicate as DuplicateException;
use Utopia\Database\Helpers\Permission;
use Utopia\Database\Helpers\Role;
use Utopia\Database\Query;
use Utopia\Logger\Log;
use Utopia\Messaging\Adapter\Email as EmailAdapter;
use Utopia\Messaging\Adapter\Email\SMTP;
use Utopia\Messaging\Messages\Email as EmailMessage;
use Utopia\Messaging\Messages\Email\Attachment;
use Utopia\Platform\Action;
use Utopia\Queue\Message;
use Utopia\Registry\Registry;
use Utopia\System\System;
class Notifications extends Action
{
protected int $previewMaxLen = 150;
protected string $whitespaceCodes = '&#xa0;&#x200C;&#x200B;&#x200D;&#x200E;&#x200F;&#xFEFF;';
/**
* Tracking pixel JWT lifetime: 30 days.
*/
private const TRACKING_JWT_TTL = 2592000;
/**
* @var array<string, string>
*/
protected array $richTextParams = [
'b' => '<strong>',
'/b' => '</strong>',
];
public static function getName(): string
{
return 'notifications';
}
public function __construct()
{
$this
->desc('Notifications worker')
->inject('message')
->inject('project')
->inject('register')
->inject('dbForProject')
->inject('log')
->callback($this->action(...));
}
public function action(Message $message, Document $project, Registry $register, Database $dbForProject, Log $log): void
{
if (\class_exists(Runtime::class)) {
Runtime::setHookFlags(SWOOLE_HOOK_ALL ^ SWOOLE_HOOK_TCP);
}
$payload = $message->getPayload();
if (empty($payload)) {
throw new Exception('Missing payload');
}
$deduplicationKey = $payload['deduplicationKey'] ?? '';
$messageId = $deduplicationKey !== '' ? \md5($deduplicationKey) : '';
if ($messageId !== '' && $this->alreadyDelivered($dbForProject, $messageId)) {
$log->addTag('dedup', 'hit');
return;
}
$recipients = $this->resolveRecipients($payload);
if (empty($recipients)) {
throw new Exception('No recipients in payload');
}
foreach ($recipients as $recipient) {
$channel = $recipient['channel'];
try {
$alertId = $this->dispatch($recipient, $messageId, $payload, $project, $register, $dbForProject, $log);
if ($messageId !== '' && $channel === NOTIFICATION_TYPE_WEBHOOK && $alertId === null) {
$this->persistAlert($dbForProject, $messageId, $recipient, $payload);
}
} catch (Throwable $error) {
$log->addTag('channel', $channel);
$log->addTag('error', $error->getMessage());
throw $error;
}
}
}
/**
* @return array<int, array{address: string, channel: string, signatureKey?: string, userId?: string, teamId?: string}>
*/
private function resolveRecipients(array $payload): array
{
$recipients = $payload['recipients'] ?? [];
if (!empty($recipients)) {
return $recipients;
}
$address = $payload['recipient'] ?? '';
if ($address === '') {
return [];
}
return [['address' => $address, 'channel' => NOTIFICATION_TYPE_EMAIL]];
}
/**
* Look up an existing alert by the indexed `messageId` attribute.
*
* Greptile P1 #1: persistAlert and the Console adapter both write
* compound `$id`s (messageId + recipient hash), so a direct
* `getDocument($messageId)` would always miss. Query the attribute.
*/
private function alreadyDelivered(Database $database, string $messageId): bool
{
try {
$matches = $database->find('alerts', [
Query::equal('messageId', [$messageId]),
Query::limit(1),
]);
return !empty($matches);
} catch (Throwable) {
return false;
}
}
/**
* Dispatch a single recipient through the channel-appropriate adapter.
*
* Returns the alertId when the dispatcher (or its adapter) has already
* persisted an alert row, so the action loop knows to skip persistence.
* Returns null when persistence is the caller's responsibility.
*
* @param array{address: string, channel: string, signatureKey?: string, userId?: string, teamId?: string} $recipient
*/
protected function dispatch(
array $recipient,
string $messageId,
array $payload,
Document $project,
Registry $register,
Database $database,
Log $log,
): ?string {
$channel = $recipient['channel'];
return match ($channel) {
NOTIFICATION_TYPE_EMAIL => $this->dispatchEmail($recipient, $messageId, $payload, $project, $register, $database, $log),
NOTIFICATION_TYPE_CONSOLE => $this->dispatchConsole($recipient, $messageId, $payload, $database),
NOTIFICATION_TYPE_WEBHOOK => $this->dispatchWebhook($recipient, $payload, $log),
default => throw new Exception('Unsupported notification channel: ' . $channel),
};
}
/**
* @param array{address: string, channel: string, signatureKey?: string, userId?: string, teamId?: string} $recipient
*/
protected function dispatchEmail(
array $recipient,
string $messageId,
array $payload,
Document $project,
Registry $register,
Database $database,
Log $log,
): ?string {
$address = $recipient['address'];
$smtp = $this->resolveSmtpConfig($project);
if (empty($smtp) && empty(System::getEnv('_APP_SMTP_HOST'))) {
throw new Exception('Skipped email notification. No SMTP configuration has been set.');
}
$type = empty($smtp) ? 'cloud' : 'smtp';
$log->addTag('type', $type);
$protocol = System::getEnv('_APP_OPTIONS_FORCE_HTTPS', 'disabled') === 'disabled' ? 'http' : 'https';
$consoleHostname = System::getEnv('_APP_CONSOLE_DOMAIN', System::getEnv('_APP_DOMAIN', 'localhost'));
$subject = $payload['subject'] ?? '';
$variables = $payload['variables'] ?? [];
$variables = \array_merge($variables, $payload['templateParams'] ?? []);
$variables['host'] = $protocol . '://' . $consoleHostname;
$name = $payload['name'] ?? '';
$body = $payload['body'] ?? '';
$preview = $payload['preview'] ?? '';
$variables['subject'] = $subject;
$variables['heading'] = $variables['heading'] ?? $subject;
$variables['year'] = \date('Y');
$attachment = $payload['attachment'] ?? [];
$bodyTemplate = $payload['bodyTemplate'] ?? '';
if (empty($bodyTemplate)) {
$bodyTemplate = $payload['template'] ?? '';
}
if (empty($bodyTemplate)) {
$bodyTemplate = __DIR__ . '/../../../../app/config/locale/templates/email-base.tpl';
}
$bodyTemplate = Template::fromFile($bodyTemplate);
$bodyTemplate->setParam('{{body}}', $body, escapeHtml: false);
foreach ($variables as $key => $value) {
$bodyTemplate->setParam('{{' . $key . '}}', $value, escapeHtml: $key !== 'redirect');
}
foreach ($this->richTextParams as $key => $value) {
$bodyTemplate->setParam('{{' . $key . '}}', $value, escapeHtml: false);
}
$previewWhitespace = '';
if (!empty($preview)) {
$previewTemplate = Template::fromString($preview);
foreach ($variables as $key => $value) {
$previewTemplate->setParam('{{' . $key . '}}', $value);
}
$preview = \strip_tags($previewTemplate->render());
$previewLen = \strlen($preview);
if ($previewLen < $this->previewMaxLen) {
$previewWhitespace = \str_repeat($this->whitespaceCodes, $this->previewMaxLen - $previewLen);
}
}
$bodyTemplate->setParam('{{preview}}', $preview);
$bodyTemplate->setParam('{{previewWhitespace}}', $previewWhitespace, false);
$body = $bodyTemplate->render();
$subjectTemplate = Template::fromString($subject);
foreach ($variables as $key => $value) {
$subjectTemplate->setParam('{{' . $key . '}}', $value);
}
$subject = \strip_tags($subjectTemplate->render());
// Persist alert BEFORE adapter send so the alertId is available for
// the tracking pixel. Failure to persist still allows the email to
// go out unsignals (we degrade gracefully).
$alertId = null;
if ($messageId !== '') {
$alertId = $this->persistAlert($database, $messageId, $recipient, $payload);
}
// C3 tracking pixel: only injectable when we have a userId AND a
// persisted alertId AND a signing key.
$userId = $recipient['userId'] ?? '';
$opensslKey = System::getEnv('_APP_OPENSSL_KEY_V1');
if ($alertId !== null && $userId !== '' && !empty($opensslKey)) {
$body = $this->injectTrackingPixel($body, $alertId, $userId, $opensslKey);
}
/** @var EmailAdapter $adapter */
$adapter = empty($smtp)
? $register->get('smtp')
: new SMTP(
host: $smtp['host'],
port: (int) $smtp['port'],
username: $smtp['username'] ?? '',
password: $smtp['password'] ?? '',
smtpSecure: $smtp['secure'] ?? '',
smtpAutoTLS: false,
xMailer: 'Appwrite Mailer',
timeout: 10,
keepAlive: true,
timelimit: 30,
);
$defaultFromEmail = System::getEnv('_APP_SYSTEM_EMAIL_ADDRESS', APP_EMAIL_TEAM);
$defaultFromName = \urldecode(System::getEnv('_APP_SYSTEM_EMAIL_NAME', APP_NAME . ' Server'));
$fromEmail = !empty($smtp) ? ($smtp['senderEmail'] ?? $defaultFromEmail) : $defaultFromEmail;
$fromName = !empty($smtp) ? ($smtp['senderName'] ?? $defaultFromName) : $defaultFromName;
$replyTo = $defaultFromEmail;
$replyToName = $defaultFromName;
if (!empty($smtp)) {
$smtpReplyToEmail = $smtp['replyToEmail'] ?? $smtp['replyTo'] ?? '';
$replyTo = !empty($smtpReplyToEmail) ? $smtpReplyToEmail : ($smtp['senderEmail'] ?? $replyTo);
$replyToName = !empty($smtp['replyToName']) ? $smtp['replyToName'] : ($smtp['senderName'] ?? $replyToName);
}
$attachments = null;
if (!empty($attachment['content'] ?? '')) {
$attachments = [
new Attachment(
name: $attachment['filename'] ?? 'unknown.file',
path: '',
type: $attachment['type'] ?? 'plain/text',
content: \base64_decode($attachment['content']),
),
];
}
$emailMessage = new EmailMessage(
to: [['email' => $address, 'name' => $name]],
subject: $subject,
content: $body,
fromName: $fromName,
fromEmail: $fromEmail,
replyToName: $replyToName,
replyToEmail: $replyTo,
attachments: $attachments,
html: true,
);
try {
$adapter->send($emailMessage);
} catch (Throwable $error) {
if ($type === 'smtp') {
throw new Exception('Error sending notification: ' . $error->getMessage(), 401);
}
throw new Exception('Error sending notification: ' . $error->getMessage(), 500);
}
return $alertId;
}
/**
* @param array{address: string, channel: string, signatureKey?: string, userId?: string, teamId?: string} $recipient
*/
protected function dispatchConsole(array $recipient, string $messageId, array $payload, Database $database): ?string
{
$project = $payload['project'] ?? null;
$projectId = \is_array($project) ? ($project['$id'] ?? null) : null;
$title = $payload['subject'] ?? '';
$body = $payload['body'] ?? '';
$params = $payload['templateParams'] ?? ($payload['variables'] ?? []);
if ($title !== '' && !empty($params)) {
$rendered = Template::fromString($title);
foreach ($params as $key => $value) {
$rendered->setParam('{{' . $key . '}}', (string) $value);
}
$title = \strip_tags($rendered->render());
}
$userId = $recipient['userId'] ?? $recipient['address'];
$teamId = $recipient['teamId'] ?? '';
$consoleRecipient = [];
if ($userId !== '') {
$consoleRecipient['userId'] = $userId;
}
if ($teamId !== '') {
$consoleRecipient['teamId'] = $teamId;
}
$consoleMessage = new ConsoleMessage(
recipients: [$consoleRecipient],
title: $title,
body: $body,
type: $payload['type'] ?? 'info',
messageId: $messageId !== '' ? $messageId : null,
projectId: $projectId,
);
$adapter = new ConsoleAdapter($database);
$result = $adapter->send($consoleMessage);
// Greptile P1 #4: surface adapter failures. The Console adapter
// catches per-recipient exceptions and reports zero deliveries via
// `deliveredTo`; without this throw the worker would silently
// succeed on a hard write failure.
if (($result['deliveredTo'] ?? 0) === 0) {
$error = $result['results'][0]['error'] ?? 'unknown error';
throw new Exception('Console alert delivery failed: ' . $error);
}
// Adapter persisted the alert, so the action loop must NOT
// call persistAlert again (Greptile P1 #3).
return null;
}
/**
* @param array{address: string, channel: string, signatureKey?: string, userId?: string, teamId?: string} $recipient
*/
protected function dispatchWebhook(array $recipient, array $payload, Log $log): ?string
{
$address = $recipient['address'];
$signatureKey = $recipient['signatureKey'] ?? null;
$body = [
'subject' => $payload['subject'] ?? '',
'body' => $payload['body'] ?? '',
'template' => $payload['template'] ?? '',
'params' => $payload['templateParams'] ?? [],
'project' => \is_array($payload['project'] ?? null) ? ($payload['project']['$id'] ?? null) : null,
'deduplicationKey' => $payload['deduplicationKey'] ?? '',
'events' => $payload['events'] ?? [],
];
if ($signatureKey === null || $signatureKey === '') {
$log->addTag('webhook_signed', 'false');
}
$message = new WebhookMessage(
urls: [$address],
payload: $body,
signingSecret: $signatureKey,
);
$adapter = new WebhookAdapter();
$result = $adapter->send($message);
if (($result['deliveredTo'] ?? 0) === 0) {
$error = $result['results'][0]['error'] ?? 'Unknown error';
throw new Exception('Webhook delivery failed: ' . $error);
}
// Caller persists the alert AFTER successful dispatch.
return null;
}
/**
* Persist an alert row. Returns the alertId so callers can build a
* tracking-pixel URL or otherwise reference the row.
*
* Idempotent: on a duplicate composite-key violation the existing
* row's id is returned.
*
* @param array{address: string, channel: string, signatureKey?: string, userId?: string, teamId?: string} $recipient
*/
protected function persistAlert(Database $database, string $messageId, array $recipient, array $payload): string
{
$project = $payload['project'] ?? null;
$projectId = \is_array($project) ? ($project['$id'] ?? null) : null;
$channel = $recipient['channel'];
$address = $recipient['address'];
$userId = $recipient['userId'] ?? '';
$teamId = $recipient['teamId'] ?? '';
// Console alerts derive userId from address when no explicit
// userId is supplied (matches Console adapter's own bookkeeping).
if ($channel === NOTIFICATION_TYPE_CONSOLE && $userId === '' && $teamId === '') {
$userId = $address;
}
$idSuffix = \substr(\md5($channel . ':' . $address . ':' . $userId . ':' . $teamId), 0, 8);
$alertId = $messageId . '_' . $idSuffix;
$permissions = $this->buildAlertPermissions($userId, $teamId);
if (empty($permissions)) {
$permissions = $payload['permissions'] ?? [];
}
$document = new Document([
'$id' => $alertId,
'$permissions' => $permissions,
'messageId' => $messageId,
'type' => $payload['type'] ?? 'info',
'channel' => $channel,
'userId' => $userId !== '' ? $userId : null,
'teamId' => $teamId !== '' ? $teamId : null,
'projectId' => $projectId,
'title' => $payload['subject'] ?? '',
'body' => $payload['body'] ?? '',
'read' => false,
]);
try {
$database->createDocument('alerts', $document);
return $alertId;
} catch (DuplicateException) {
$existing = $database->getDocument('alerts', $alertId);
return $existing->isEmpty() ? $alertId : $existing->getId();
}
}
/**
* @return array<string>
*/
private function buildAlertPermissions(string $userId, string $teamId): array
{
$permissions = [];
if ($userId !== '') {
$permissions[] = Permission::read(Role::user($userId));
$permissions[] = Permission::update(Role::user($userId));
$permissions[] = Permission::delete(Role::user($userId));
}
if ($teamId !== '') {
$permissions[] = Permission::read(Role::team($teamId));
$permissions[] = Permission::update(Role::team($teamId, 'owner'));
$permissions[] = Permission::delete(Role::team($teamId, 'owner'));
}
return $permissions;
}
/**
* Resolve project SMTP config to the wire shape Mails.php expects.
* ST4 stripped `smtp` and `customMailOptions` from the Notification
* event payload, so the worker now reads from the project Document.
* Falls back to env-driven cloud SMTP when the project has not
* configured custom SMTP.
*
* @return array<string, mixed>
*/
private function resolveSmtpConfig(Document $project): array
{
$smtp = $project->getAttribute('smtp', []);
if (!\is_array($smtp) || empty($smtp['enabled'] ?? false)) {
return [];
}
return [
'host' => $smtp['host'] ?? '',
'port' => $smtp['port'] ?? '',
'username' => $smtp['username'] ?? '',
'password' => $smtp['password'] ?? '',
'secure' => $smtp['secure'] ?? '',
'senderEmail' => $smtp['senderEmail'] ?? '',
'senderName' => $smtp['senderName'] ?? '',
'replyToEmail' => $smtp['replyToEmail'] ?? $smtp['replyTo'] ?? '',
'replyToName' => $smtp['replyToName'] ?? '',
];
}
/**
* Splice a 1x1 tracking pixel before the last `</body>` tag (or
* append at the end if the body has no closing tag). The pixel
* carries a 30-day JWT identifying the alert and user, which the
* `/v1/account/alerts/:alertId/track` endpoint verifies before
* marking the alert as read.
*/
private function injectTrackingPixel(string $body, string $alertId, string $userId, string $opensslKey): string
{
$jwt = (new JWT($opensslKey, 'HS256', self::TRACKING_JWT_TTL, 0))
->encode([
'alertId' => $alertId,
'userId' => $userId,
]);
$protocol = System::getEnv('_APP_OPTIONS_FORCE_HTTPS', 'disabled') === 'disabled' ? 'http' : 'https';
$hostname = System::getEnv('_APP_DOMAIN', 'localhost');
$pixelUrl = $protocol . '://' . $hostname . '/v1/account/alerts/' . $alertId . '/track?jwt=' . \urlencode($jwt);
$pixel = '<img src="' . \htmlspecialchars($pixelUrl, ENT_QUOTES, 'UTF-8') . '" width="1" height="1" alt="" style="display:none" />';
// Case-insensitive splice before the LAST </body>.
if (\preg_match('/<\/body\s*>(?!.*<\/body\s*>)/is', $body)) {
return \preg_replace('/<\/body\s*>(?!.*<\/body\s*>)/is', $pixel . '$0', $body, 1) ?? ($body . $pixel);
}
return $body . $pixel;
}
}
+60 -20
View File
@@ -2,8 +2,8 @@
namespace Appwrite\Platform\Workers;
use Appwrite\Event\Mail;
use Appwrite\Event\Message\Usage as UsageMessage;
use Appwrite\Event\Notification;
use Appwrite\Event\Publisher\Usage as UsagePublisher;
use Appwrite\Template\Template;
use Appwrite\Usage\Context as UsageContext;
@@ -36,7 +36,7 @@ class Webhooks extends Action
->inject('message')
->inject('project')
->inject('dbForPlatform')
->inject('queueForMails')
->inject('queueForNotifications')
->inject('publisherForUsage')
->inject('log')
->inject('plan')
@@ -47,14 +47,14 @@ class Webhooks extends Action
* @param Message $message
* @param Document $project
* @param Database $dbForPlatform
* @param Mail $queueForMails
* @param Notification $queueForNotifications
* @param UsagePublisher $publisherForUsage
* @param Log $log
* @param array $plan
* @return void
* @throws Exception
*/
public function action(Message $message, Document $project, Database $dbForPlatform, Mail $queueForMails, UsagePublisher $publisherForUsage, Log $log, array $plan): void
public function action(Message $message, Document $project, Database $dbForPlatform, Notification $queueForNotifications, UsagePublisher $publisherForUsage, Log $log, array $plan): void
{
$this->errors = [];
$payload = $message->getPayload();
@@ -73,7 +73,7 @@ class Webhooks extends Action
foreach ($project->getAttribute('webhooks', []) as $webhook) {
if (array_intersect($webhook->getAttribute('events', []), $events)) {
$this->execute($events, $webhookPayload, $webhook, $user, $project, $dbForPlatform, $queueForMails, $publisherForUsage, $plan);
$this->execute($events, $webhookPayload, $webhook, $user, $project, $dbForPlatform, $queueForNotifications, $publisherForUsage, $plan);
}
}
@@ -89,11 +89,11 @@ class Webhooks extends Action
* @param Document $user
* @param Document $project
* @param Database $dbForPlatform
* @param Mail $queueForMails
* @param Notification $queueForNotifications
* @param array $plan
* @return void
*/
private function execute(array $events, string $payload, Document $webhook, Document $user, Document $project, Database $dbForPlatform, Mail $queueForMails, UsagePublisher $publisherForUsage, array $plan): void
private function execute(array $events, string $payload, Document $webhook, Document $user, Document $project, Database $dbForPlatform, Notification $queueForNotifications, UsagePublisher $publisherForUsage, array $plan): void
{
if ($webhook->getAttribute('enabled') !== true) {
return;
@@ -171,7 +171,7 @@ class Webhooks extends Action
if ($attempts >= \intval(System::getEnv('_APP_WEBHOOK_MAX_FAILED_ATTEMPTS', '10'))) {
$webhook->setAttribute('enabled', false);
$updatePayload['enabled'] = false;
$this->sendEmailAlert($attempts, $statusCode, $webhook, $project, $dbForPlatform, $queueForMails, $plan);
$this->sendAlert($attempts, $statusCode, $webhook, $project, $dbForPlatform, $queueForNotifications, $plan);
}
$dbForPlatform->updateDocument('webhooks', $webhook->getId(), new Document($updatePayload));
@@ -203,26 +203,49 @@ class Webhooks extends Action
* @param Document $webhook
* @param Document $project
* @param Database $dbForPlatform
* @param Mail $queueForMails
* @param Notification $queueForNotifications
* @param array $plan
* @return void
*/
public function sendEmailAlert(int $attempts, mixed $statusCode, Document $webhook, Document $project, Database $dbForPlatform, Mail $queueForMails, array $plan): void
public function sendAlert(int $attempts, mixed $statusCode, Document $webhook, Document $project, Database $dbForPlatform, Notification $queueForNotifications, array $plan): void
{
$memberships = $dbForPlatform->find('memberships', [
Query::equal('teamInternalId', [$project->getAttribute('teamInternalId')]),
Query::limit(APP_LIMIT_SUBQUERY)
]);
$userIds = array_column(\array_map(fn ($membership) => $membership->getArrayCopy(), $memberships), 'userId');
// Webhook-paused alerts go only to project owners — non-owner team members do not receive them.
$ownerMemberships = \array_filter(
$memberships,
fn (Document $membership) => \in_array('owner', $membership->getAttribute('roles', []), true)
);
if (empty($ownerMemberships)) {
return;
}
$userIds = \array_values(\array_unique(\array_filter(\array_map(
fn (Document $membership) => $membership->getAttribute('userId'),
$ownerMemberships
))));
if (empty($userIds)) {
return;
}
$users = $dbForPlatform->find('users', [
Query::equal('$id', $userIds),
Query::limit(APP_LIMIT_SUBQUERY),
]);
if (empty($users)) {
return;
}
$projectId = $project->getId();
$region = $project->getAttribute('region', 'default');
$webhookId = $webhook->getId();
$teamId = $project->getAttribute('teamId');
$template = Template::fromFile(__DIR__ . '/../../../../app/config/locale/templates/email-webhook-failed.tpl');
@@ -241,7 +264,6 @@ class Webhooks extends Action
$template->setParam('{{termsUrl}}', $plan['termsUrl'] ?? APP_EMAIL_TERMS_URL);
$template->setParam('{{privacyUrl}}', $plan['privacyUrl'] ?? APP_EMAIL_PRIVACY_URL);
// TODO: Use setbodyTemplate once #7307 is merged
$subject = 'Webhook deliveries have been paused';
$preview = 'Webhook deliveries to your endpoint have been paused.';
$body = Template::fromFile(__DIR__ . '/../../../../app/config/locale/templates/email-base-styled.tpl');
@@ -249,20 +271,38 @@ class Webhooks extends Action
$body
->setParam('{{subject}}', $subject)
->setParam('{{message}}', $template->render())
->setParam('{{year}}', date("Y"));
->setParam('{{year}}', date('Y'));
$queueForMails
$queueForNotifications
->setProject($project)
->setSubject($subject)
->setPreview($preview)
->setBody($body->render());
->setBody($body->render())
->setDeduplicationKey('webhook:' . $webhook->getId() . ':paused:' . $attempts);
foreach ($users as $user) {
$queueForMails
->setVariables(['user' => $user->getAttribute('name', '')])
->setName($user->getAttribute('name', ''))
->setRecipient($user->getAttribute('email'))
->trigger();
$email = $user->getAttribute('email');
$userId = $user->getId();
if (!empty($email)) {
$queueForNotifications->addRecipient(
$email,
NOTIFICATION_TYPE_EMAIL,
null,
$userId,
$teamId,
);
}
$queueForNotifications->addRecipient(
$userId,
NOTIFICATION_TYPE_CONSOLE,
null,
$userId,
$teamId,
);
}
$queueForNotifications->trigger();
}
}
@@ -0,0 +1,23 @@
<?php
namespace Appwrite\Utopia\Database\Validator\Queries;
class Alerts extends Base
{
public const ALLOWED_ATTRIBUTES = [
'read',
'type',
'channel',
'messageId',
'projectId',
];
/**
* Expression constructor
*
*/
public function __construct()
{
parent::__construct('alerts', self::ALLOWED_ATTRIBUTES);
}
}
@@ -0,0 +1,114 @@
<?php
namespace Appwrite\Utopia\Messaging\Adapter;
use Appwrite\Utopia\Messaging\Messages\Console as ConsoleMessage;
use Utopia\Database\Database;
use Utopia\Database\Document;
use Utopia\Database\Helpers\ID;
use Utopia\Database\Helpers\Permission;
use Utopia\Database\Helpers\Role;
use Utopia\Messaging\Adapter;
use Utopia\Messaging\Message;
use Utopia\Messaging\Response;
class Console extends Adapter
{
protected const NAME = 'Console';
protected const TYPE = 'console';
protected const MESSAGE_TYPE = ConsoleMessage::class;
public function __construct(protected Database $database)
{
}
public function getName(): string
{
return static::NAME;
}
public function getType(): string
{
return static::TYPE;
}
public function getMessageType(): string
{
return static::MESSAGE_TYPE;
}
public function getMaxMessagesPerRequest(): int
{
return 1000;
}
public function send(Message $message): array
{
if (!$message instanceof ConsoleMessage) {
throw new \Exception('Invalid message type.');
}
return $this->process($message);
}
protected function process(ConsoleMessage $message): array
{
$response = new Response($this->getType());
$delivered = 0;
foreach ($message->getRecipients() as $recipient) {
$userId = $recipient['userId'] ?? '';
$teamId = $recipient['teamId'] ?? '';
$key = $userId !== '' ? $userId : $teamId;
$messageId = $message->getMessageId();
$recipientKey = $userId !== '' ? 'user:' . $userId : 'team:' . $teamId;
$documentId = $messageId !== null
? $messageId . '_' . \substr(\md5($recipientKey), 0, 8)
: ID::unique();
try {
$document = new Document([
'$id' => $documentId,
'$permissions' => $this->buildPermissions($userId, $teamId),
'messageId' => $messageId,
'type' => $message->getType(),
'channel' => self::TYPE,
'userId' => $userId !== '' ? $userId : null,
'teamId' => $teamId !== '' ? $teamId : null,
'projectId' => $message->getProjectId(),
'title' => $message->getTitle(),
'body' => $message->getBody(),
]);
$this->database->createDocument('alerts', $document);
$delivered++;
$response->addResult($key);
} catch (\Throwable $error) {
$response->addResult($key, $error->getMessage());
}
}
$response->setDeliveredTo($delivered);
return $response->toArray();
}
/**
* @return array<string>
*/
private function buildPermissions(string $userId, string $teamId): array
{
$permissions = [];
if ($userId !== '') {
$permissions[] = Permission::read(Role::user($userId));
$permissions[] = Permission::update(Role::user($userId));
$permissions[] = Permission::delete(Role::user($userId));
}
if ($teamId !== '') {
$permissions[] = Permission::read(Role::team($teamId));
$permissions[] = Permission::update(Role::team($teamId, 'owner'));
$permissions[] = Permission::delete(Role::team($teamId, 'owner'));
}
return $permissions;
}
}
@@ -0,0 +1,124 @@
<?php
namespace Appwrite\Utopia\Messaging\Adapter;
use Appwrite\Utopia\Messaging\Messages\Webhook as WebhookMessage;
use Utopia\Fetch\Client as FetchClient;
use Utopia\Fetch\Exception as FetchException;
use Utopia\Messaging\Adapter;
use Utopia\Messaging\Message;
use Utopia\Messaging\Response;
class Webhook extends Adapter
{
protected const NAME = 'Webhook';
protected const TYPE = 'webhook';
protected const MESSAGE_TYPE = WebhookMessage::class;
protected const SIGNATURE_HEADER = 'X-Appwrite-Webhook-Signature';
protected const TIMESTAMP_HEADER = 'X-Appwrite-Webhook-Timestamp';
public function getName(): string
{
return static::NAME;
}
public function getType(): string
{
return static::TYPE;
}
public function getMessageType(): string
{
return static::MESSAGE_TYPE;
}
public function getMaxMessagesPerRequest(): int
{
return 100;
}
public function send(Message $message): array
{
if (!$message instanceof WebhookMessage) {
throw new \Exception('Invalid message type.');
}
return $this->process($message);
}
protected function process(WebhookMessage $message): array
{
$response = new Response($this->getType());
$body = \json_encode($message->getPayload(), JSON_THROW_ON_ERROR);
$timestamp = (string) \time();
$headers = [
'Content-Type: application/json',
self::TIMESTAMP_HEADER . ': ' . $timestamp,
];
$secret = $message->getSigningSecret();
if ($secret !== null && $secret !== '') {
$signature = \hash_hmac('sha256', $timestamp . '.' . $body, $secret);
$headers[] = self::SIGNATURE_HEADER . ': sha256=' . $signature;
}
foreach ($message->getHeaders() as $name => $value) {
$headers[] = $name . ': ' . $value;
}
$delivered = 0;
foreach ($message->getUrls() as $url) {
$result = $this->dispatch('POST', $url, $headers, $body, $message->getTimeout());
if ($result['statusCode'] >= 200 && $result['statusCode'] < 300 && empty($result['error'])) {
$delivered++;
$response->addResult($url);
} else {
$response->addResult($url, $result['error'] ?: ('HTTP ' . $result['statusCode']));
}
}
$response->setDeliveredTo($delivered);
return $response->toArray();
}
/**
* @param array<int, string> $headers
* @return array{statusCode: int, response: string|null, error: string|null}
*/
protected function dispatch(string $method, string $url, array $headers, string $body, int $timeout): array
{
$client = new FetchClient();
$client
->setTimeout($timeout * 1000)
->setConnectTimeout(\min(10, $timeout) * 1000)
->setAllowRedirects(false)
->setUserAgent('Appwrite Webhook');
foreach ($headers as $header) {
$parts = \explode(':', $header, 2);
if (\count($parts) === 2) {
$client->addHeader(\trim($parts[0]), \trim($parts[1]));
}
}
try {
$response = $client->fetch($url, $method, $body);
} catch (FetchException $exception) {
return [
'statusCode' => 0,
'response' => null,
'error' => $exception->getMessage(),
];
}
$output = $response->getBody();
return [
'statusCode' => $response->getStatusCode(),
'response' => \is_string($output) ? $output : null,
'error' => null,
];
}
}
@@ -0,0 +1,62 @@
<?php
namespace Appwrite\Utopia\Messaging\Messages;
use Utopia\Messaging\Message;
class Console implements Message
{
/**
* @param array<int, array{userId?: string, teamId?: string}> $recipients
*/
public function __construct(
protected array $recipients,
protected string $title,
protected string $body,
protected string $type = 'info',
protected ?string $messageId = null,
protected ?string $projectId = null,
) {
}
/**
* @return array<int, array{userId?: string, teamId?: string}>
*/
public function getRecipients(): array
{
return $this->recipients;
}
/**
* @return array<int, array{userId?: string, teamId?: string}>
*/
public function getTo(): array
{
return $this->recipients;
}
public function getTitle(): string
{
return $this->title;
}
public function getBody(): string
{
return $this->body;
}
public function getType(): string
{
return $this->type;
}
public function getMessageId(): ?string
{
return $this->messageId;
}
public function getProjectId(): ?string
{
return $this->projectId;
}
}
@@ -0,0 +1,66 @@
<?php
namespace Appwrite\Utopia\Messaging\Messages;
use Utopia\Messaging\Message;
class Webhook implements Message
{
/**
* @param array<int, string> $urls
* @param array<string, mixed> $payload
* @param array<string, string> $headers
*/
public function __construct(
protected array $urls,
protected array $payload,
protected ?string $signingSecret = null,
protected array $headers = [],
protected int $timeout = 30,
) {
}
/**
* @return array<int, string>
*/
public function getUrls(): array
{
return $this->urls;
}
/**
* Alias used by the base adapter to bound max messages per request.
*
* @return array<int, string>
*/
public function getTo(): array
{
return $this->urls;
}
/**
* @return array<string, mixed>
*/
public function getPayload(): array
{
return $this->payload;
}
public function getSigningSecret(): ?string
{
return $this->signingSecret;
}
/**
* @return array<string, string>
*/
public function getHeaders(): array
{
return $this->headers;
}
public function getTimeout(): int
{
return $this->timeout;
}
}
+4
View File
@@ -67,6 +67,10 @@ class Response extends SwooleResponse
public const MODEL_ROW = 'row';
public const MODEL_ROW_LIST = 'rowList';
// Alerts
public const MODEL_ALERT = 'alert';
public const MODEL_ALERT_LIST = 'alertList';
// Database Attributes
public const MODEL_ATTRIBUTE = 'attribute';
public const MODEL_ATTRIBUTE_LIST = 'attributeList';
@@ -0,0 +1,112 @@
<?php
namespace Appwrite\Utopia\Response\Model;
use Appwrite\Utopia\Response;
use Appwrite\Utopia\Response\Model;
class Alert extends Model
{
public function __construct()
{
$this
->addRule('$id', [
'type' => self::TYPE_STRING,
'description' => 'Alert ID.',
'default' => '',
'example' => '5e5ea5c16897e',
])
->addRule('$createdAt', [
'type' => self::TYPE_DATETIME,
'description' => 'Alert creation date in ISO 8601 format.',
'default' => '',
'example' => self::TYPE_DATETIME_EXAMPLE,
])
->addRule('$updatedAt', [
'type' => self::TYPE_DATETIME,
'description' => 'Alert update date in ISO 8601 format.',
'default' => '',
'example' => self::TYPE_DATETIME_EXAMPLE,
])
->addRule('messageId', [
'type' => self::TYPE_STRING,
'description' => 'Stable message ID used for dedup.',
'default' => '',
'example' => 'session.create',
'required' => false,
])
->addRule('type', [
'type' => self::TYPE_STRING,
'description' => 'Alert type: info, warning, error.',
'default' => 'info',
'example' => 'info',
])
->addRule('channel', [
'type' => self::TYPE_STRING,
'description' => 'Channel: email, sms, push, console, webhook.',
'default' => '',
'example' => 'email',
])
->addRule('userId', [
'type' => self::TYPE_STRING,
'description' => 'User this alert is addressed to.',
'default' => '',
'example' => '5e5bb8c16897e',
'required' => false,
])
->addRule('teamId', [
'type' => self::TYPE_STRING,
'description' => 'Team this alert is addressed to.',
'default' => '',
'example' => '5e5bb8c16897e',
'required' => false,
])
->addRule('projectId', [
'type' => self::TYPE_STRING,
'description' => 'Project the alert pertains to.',
'default' => '',
'example' => '5e5bb8c16897e',
'required' => false,
])
->addRule('title', [
'type' => self::TYPE_STRING,
'description' => 'Alert title.',
'default' => '',
'example' => 'New sign-in detected',
])
->addRule('body', [
'type' => self::TYPE_STRING,
'description' => 'Alert body.',
'default' => '',
'example' => 'A new device signed in to your account.',
])
->addRule('read', [
'type' => self::TYPE_BOOLEAN,
'description' => 'Whether the alert has been read.',
'default' => false,
'example' => false,
'required' => false,
])
;
}
/**
* Get Name
*
* @return string
*/
public function getName(): string
{
return 'Alert';
}
/**
* Get Type
*
* @return string
*/
public function getType(): string
{
return Response::MODEL_ALERT;
}
}
@@ -0,0 +1,439 @@
<?php
namespace Tests\E2E\Services\Notifications;
use Ahc\Jwt\JWT;
use Tests\E2E\Client;
use Utopia\Database\Helpers\ID;
use Utopia\System\System;
/**
* End-to-end coverage for the notifications queue health surface and the
* account-alerts user-facing API.
*
* The notification worker itself is exercised in unit tests with a pinned
* queue payload — the server side cannot deterministically inject a
* Notification onto the live queue without an admin endpoint, so the health
* portion validates the public contract that ops and KEDA scale on:
*
* - GET /v1/health/queue/notifications returns the live queue depth
* - the threshold guard returns 503 when the depth exceeds the budget
* - the failed-jobs surface accepts the notifications queue name
*
* The alerts portion exercises the full webhook-paused fanout end-to-end:
*
* - GET /v1/account/alerts (empty + populated)
* - PATCH /v1/account/alerts/:alertId/read (happy + unauthorized)
* - GET /v1/account/alerts/:alertId/track (valid JWT + invalid JWT)
*
* Dedup, per-channel dispatch, and webhook signing are covered by:
* - tests/unit/Platform/Workers/NotificationsTest.php
* - tests/unit/Utopia/Messaging/Adapter/ConsoleTest.php
* - tests/unit/Utopia/Messaging/Adapter/WebhookTest.php
*/
trait NotificationsBase
{
public function testHealthQueueNotificationsReportsSize(): void
{
$response = $this->client->call(Client::METHOD_GET, '/health/queue/notifications', \array_merge([
'content-type' => 'application/json',
'x-appwrite-project' => $this->getProject()['$id'],
], $this->getHeaders()));
$this->assertSame(200, $response['headers']['status-code']);
$this->assertIsInt($response['body']['size']);
$this->assertGreaterThanOrEqual(0, $response['body']['size']);
}
public function testHealthQueueNotificationsThresholdGuard(): void
{
$response = $this->client->call(Client::METHOD_GET, '/health/queue/notifications', \array_merge([
'content-type' => 'application/json',
'x-appwrite-project' => $this->getProject()['$id'],
], $this->getHeaders()), ['threshold' => '0']);
$this->assertContains($response['headers']['status-code'], [200, 503]);
}
public function testHealthQueueFailedAcceptsNotifications(): void
{
$response = $this->client->call(Client::METHOD_GET, '/health/queue/failed/v1-notifications', \array_merge([
'content-type' => 'application/json',
'x-appwrite-project' => $this->getProject()['$id'],
], $this->getHeaders()));
$this->assertSame(200, $response['headers']['status-code']);
$this->assertIsInt($response['body']['size']);
}
public function testListAccountAlertsEmpty(): void
{
// Always read alerts as the console-authenticated owner of the team.
// The /v1/account/alerts endpoint is platform-scoped (dbForPlatform) and
// requires a session — server-mode API keys do not satisfy it.
$response = $this->client->call(Client::METHOD_GET, '/account/alerts', $this->getConsoleAlertHeaders());
$this->assertSame(200, $response['headers']['status-code']);
$this->assertArrayHasKey('alerts', $response['body']);
$this->assertArrayHasKey('total', $response['body']);
$this->assertIsArray($response['body']['alerts']);
$this->assertIsInt($response['body']['total']);
// The shared root console user may carry alerts from prior tests in the
// same suite — assert only that the response shape is correct and that
// counts agree.
$this->assertSame(\count($response['body']['alerts']), \min(\count($response['body']['alerts']), $response['body']['total']));
}
public function testWebhookFailureCreatesConsoleAlert(): void
{
$alertId = $this->seedWebhookFailureAlert();
$this->assertNotEmpty($alertId);
$list = $this->client->call(Client::METHOD_GET, '/account/alerts', $this->getConsoleAlertHeaders());
$this->assertSame(200, $list['headers']['status-code']);
$found = null;
foreach ($list['body']['alerts'] as $alert) {
if ($alert['$id'] === $alertId) {
$found = $alert;
break;
}
}
$this->assertNotNull($found, 'Seeded alert not present in /account/alerts response.');
$this->assertSame('console', $found['channel']);
$this->assertStringContainsStringIgnoringCase('webhook', $found['title']);
// Cache the seeded alert id for downstream tests in the same process.
self::$seededAlertId = $alertId;
}
public function testMarkAlertReadTogglesFlag(): void
{
$alertId = self::$seededAlertId ?? $this->seedWebhookFailureAlert();
$this->assertNotEmpty($alertId);
$patch = $this->client->call(
Client::METHOD_PATCH,
'/account/alerts/' . $alertId . '/read',
$this->getConsoleAlertHeaders(),
[]
);
$this->assertSame(200, $patch['headers']['status-code']);
$this->assertSame($alertId, $patch['body']['$id']);
$this->assertTrue($patch['body']['read']);
$list = $this->client->call(Client::METHOD_GET, '/account/alerts', $this->getConsoleAlertHeaders());
$this->assertSame(200, $list['headers']['status-code']);
$found = null;
foreach ($list['body']['alerts'] as $alert) {
if ($alert['$id'] === $alertId) {
$found = $alert;
break;
}
}
$this->assertNotNull($found);
$this->assertTrue($found['read']);
self::$seededAlertId = null; // alert is read — downstream tests will seed fresh
}
public function testMarkAlertReadUnauthorized(): void
{
$alertId = $this->seedWebhookFailureAlert();
$this->assertNotEmpty($alertId);
// Create a stranger console user with their own session.
$stranger = $this->createConsoleUser();
$unauthorized = $this->client->call(
Client::METHOD_PATCH,
'/account/alerts/' . $alertId . '/read',
[
'origin' => 'http://localhost',
'content-type' => 'application/json',
'cookie' => 'a_session_console=' . $stranger['session'],
'x-appwrite-project' => 'console',
'x-appwrite-mode' => 'admin',
],
[]
);
$this->assertSame(401, $unauthorized['headers']['status-code']);
$this->assertSame('user_unauthorized', $unauthorized['body']['type'] ?? '');
// Owner re-fetches — alert must still be unread.
$list = $this->client->call(Client::METHOD_GET, '/account/alerts', $this->getConsoleAlertHeaders());
$this->assertSame(200, $list['headers']['status-code']);
$found = null;
foreach ($list['body']['alerts'] as $alert) {
if ($alert['$id'] === $alertId) {
$found = $alert;
break;
}
}
$this->assertNotNull($found);
$this->assertFalse($found['read']);
self::$seededAlertId = $alertId;
}
public function testTrackingPixelTogglesRead(): void
{
$alertId = self::$seededAlertId ?? $this->seedWebhookFailureAlert();
$this->assertNotEmpty($alertId);
$secret = System::getEnv('_APP_OPENSSL_KEY_V1') ?: 'your-secret-key';
$userId = $this->getRoot()['$id'];
$jwt = (new JWT($secret, 'HS256', 2592000, 0))->encode([
'alertId' => $alertId,
'userId' => $userId,
]);
$response = $this->client->call(
Client::METHOD_GET,
'/account/alerts/' . $alertId . '/track',
['x-appwrite-project' => 'console'],
['jwt' => $jwt]
);
$this->assertSame(200, $response['headers']['status-code']);
$this->assertStringContainsString('image/png', $response['headers']['content-type']);
$this->assertNotEmpty($response['body']);
$this->assertSame("\x89PNG\r\n\x1a\n", \substr($response['body'], 0, 8), 'Response body must be a PNG.');
// Subsequent listing should report alert as read.
$list = $this->client->call(Client::METHOD_GET, '/account/alerts', $this->getConsoleAlertHeaders());
$this->assertSame(200, $list['headers']['status-code']);
$found = null;
foreach ($list['body']['alerts'] as $alert) {
if ($alert['$id'] === $alertId) {
$found = $alert;
break;
}
}
$this->assertNotNull($found);
$this->assertTrue($found['read']);
self::$seededAlertId = null;
}
public function testTrackingPixelInvalidTokenReturnsPng(): void
{
$alertId = $this->seedWebhookFailureAlert();
$this->assertNotEmpty($alertId);
$response = $this->client->call(
Client::METHOD_GET,
'/account/alerts/' . $alertId . '/track',
['x-appwrite-project' => 'console'],
['jwt' => 'tampered-or-empty']
);
$this->assertSame(200, $response['headers']['status-code']);
$this->assertStringContainsString('image/png', $response['headers']['content-type']);
$this->assertNotEmpty($response['body']);
$this->assertSame("\x89PNG\r\n\x1a\n", \substr($response['body'], 0, 8), 'Response body must be a PNG.');
// Alert must remain unread — invalid JWT is silently ignored, no DB write.
$list = $this->client->call(Client::METHOD_GET, '/account/alerts', $this->getConsoleAlertHeaders());
$this->assertSame(200, $list['headers']['status-code']);
$found = null;
foreach ($list['body']['alerts'] as $alert) {
if ($alert['$id'] === $alertId) {
$found = $alert;
break;
}
}
$this->assertNotNull($found);
$this->assertFalse($found['read']);
self::$seededAlertId = $alertId;
}
/**
* @var string|null Cached seeded alert id so consecutive tests can reuse it
* without paying the cost of another 10-failure webhook drive.
*/
protected static ?string $seededAlertId = null;
/**
* Build the auth header set used to talk to the platform-scoped
* /v1/account/alerts endpoints. Always console session, regardless of
* the trait's host (server vs console) — the endpoint requires a session
* and the root user owns the project team.
*
* @return array<string, string>
*/
protected function getConsoleAlertHeaders(): array
{
return [
'origin' => 'http://localhost',
'content-type' => 'application/json',
'cookie' => 'a_session_console=' . $this->getRoot()['session'],
'x-appwrite-project' => 'console',
'x-appwrite-mode' => 'admin',
];
}
/**
* Drive a webhook past the failure threshold so the
* Webhooks worker emits a console+email alert fanout to the project
* owner. Returns the alert id.
*
* Uses a unique webhook-per-call so concurrent tests don't share
* attempt counters.
*/
protected function seedWebhookFailureAlert(): string
{
$project = $this->getProject();
$projectId = $project['$id'];
// Register a webhook pointing at an unroutable address. The Webhook
// worker will fail every delivery and after 10 attempts pause it
// and enqueue a console+email alert for the project owner.
$webhook = $this->client->call(Client::METHOD_POST, '/webhooks', [
'origin' => 'http://localhost',
'content-type' => 'application/json',
'cookie' => 'a_session_console=' . $this->getRoot()['session'],
'x-appwrite-project' => $projectId,
'x-appwrite-mode' => 'admin',
], [
'webhookId' => ID::unique(),
'name' => 'Failing Webhook ' . \uniqid(),
'events' => ['users.*.create'],
'url' => 'http://127.0.0.1:1/',
'tls' => false,
]);
$this->assertSame(201, $webhook['headers']['status-code']);
$webhookId = $webhook['body']['$id'];
$maxAttempts = (int) System::getEnv('_APP_WEBHOOK_MAX_FAILED_ATTEMPTS', '10');
// Drive the webhook past its failure threshold by issuing user-create
// events, each of which triggers a delivery attempt the worker will
// fail. Each create event is also dispatched to the project's
// pre-existing reachable webhook — that one stays healthy.
for ($i = 0; $i < $maxAttempts + 2; $i++) {
$email = \uniqid('alert-seed-', true) . '@localhost.test';
$created = $this->client->call(Client::METHOD_POST, '/users', \array_merge([
'content-type' => 'application/json',
'x-appwrite-project' => $projectId,
], $this->getHeaders()), [
'userId' => ID::unique(),
'email' => $email,
'password' => 'password',
'name' => 'Webhook Failure Driver',
]);
// Tolerate transient 409s under parallel load.
$this->assertContains(
$created['headers']['status-code'],
[201, 409],
'User create failed while seeding webhook failure: ' . ($created['body']['message'] ?? '')
);
}
// The deduplication key (and therefore the alert's messageId hash) is
// unique per (webhook, attempts) tuple — see Webhooks worker. Compute
// the expected messageId so we can deterministically match the alert
// for *this* test instance even when other tests in the same process
// have seeded their own webhook-failure alerts.
// Driver loops max+2 events; the worker may pause anywhere from
// attempts==max to attempts==max+2 depending on which delivery
// crossed the threshold. Record possible message ids.
$expectedMessageIds = [];
for ($attempts = $maxAttempts; $attempts <= $maxAttempts + 2; $attempts++) {
$expectedMessageIds[] = \md5('webhook:' . $webhookId . ':paused:' . $attempts);
}
// Poll for the alert. Alert creation is async (notification worker)
// and webhook deliveries also queue up — give them generous time.
$alertId = null;
$this->assertEventually(function () use (&$alertId, $webhookId, $expectedMessageIds) {
$list = $this->client->call(Client::METHOD_GET, '/account/alerts', $this->getConsoleAlertHeaders());
$this->assertSame(200, $list['headers']['status-code']);
foreach ($list['body']['alerts'] as $alert) {
if (
($alert['channel'] ?? '') === 'console'
&& \in_array($alert['messageId'] ?? '', $expectedMessageIds, true)
) {
$alertId = $alert['$id'];
return;
}
}
$this->fail('No webhook-paused console alert observed yet for webhook ' . $webhookId);
}, 60000, 1000);
// Cleanup the failing webhook so it doesn't keep firing in the
// background for subsequent tests in the same process.
$this->client->call(Client::METHOD_DELETE, '/webhooks/' . $webhookId, [
'origin' => 'http://localhost',
'content-type' => 'application/json',
'cookie' => 'a_session_console=' . $this->getRoot()['session'],
'x-appwrite-project' => $projectId,
'x-appwrite-mode' => 'admin',
]);
return $alertId;
}
/**
* Create a fresh console user with its own session, separate from the
* shared root user. Used to assert that strangers cannot mark someone
* else's alert as read.
*
* @return array{$id: string, email: string, session: string}
*/
protected function createConsoleUser(): array
{
$email = \uniqid('stranger-', true) . \getmypid() . \bin2hex(\random_bytes(4)) . '@localhost.test';
$password = 'password';
$user = $this->client->call(Client::METHOD_POST, '/account', [
'origin' => 'http://localhost',
'content-type' => 'application/json',
'x-appwrite-project' => 'console',
], [
'userId' => ID::unique(),
'email' => $email,
'password' => $password,
'name' => 'Stranger',
]);
$this->assertSame(201, $user['headers']['status-code']);
$session = $this->client->call(Client::METHOD_POST, '/account/sessions/email', [
'origin' => 'http://localhost',
'content-type' => 'application/json',
'x-appwrite-project' => 'console',
], [
'email' => $email,
'password' => $password,
]);
$this->assertSame(201, $session['headers']['status-code']);
$this->assertNotEmpty($session['cookies']['a_session_console'] ?? '');
return [
'$id' => $user['body']['$id'],
'email' => $email,
'session' => $session['cookies']['a_session_console'],
];
}
}
@@ -0,0 +1,14 @@
<?php
namespace Tests\E2E\Services\Notifications;
use Tests\E2E\Scopes\ProjectCustom;
use Tests\E2E\Scopes\Scope;
use Tests\E2E\Scopes\SideConsole;
class NotificationsCustomConsoleTest extends Scope
{
use NotificationsBase;
use ProjectCustom;
use SideConsole;
}
@@ -0,0 +1,14 @@
<?php
namespace Tests\E2E\Services\Notifications;
use Tests\E2E\Scopes\ProjectCustom;
use Tests\E2E\Scopes\Scope;
use Tests\E2E\Scopes\SideServer;
class NotificationsCustomServerTest extends Scope
{
use NotificationsBase;
use ProjectCustom;
use SideServer;
}
+24
View File
@@ -292,6 +292,30 @@ services:
- _APP_SMTP_HOST
- _APP_SMTP_PORT
appwrite-worker-notifications:
entrypoint: worker-notifications
container_name: appwrite-worker-notifications
build:
context: .
restart: unless-stopped
networks:
- appwrite
depends_on:
- redis
- maildev
environment:
- _APP_ENV
- _APP_REDIS_HOST
- _APP_REDIS_PORT
- _APP_DB_HOST
- _APP_DB_PORT
- _APP_DB_SCHEMA
- _APP_DB_USER
- _APP_DB_PASS
- _APP_DB_ADAPTER
- _APP_SMTP_HOST
- _APP_SMTP_PORT
appwrite-worker-builds:
entrypoint: worker-builds
container_name: appwrite-worker-builds
@@ -0,0 +1,299 @@
<?php
namespace Tests\Unit\Platform\Workers;
use Appwrite\Platform\Workers\Notifications;
use PHPUnit\Framework\TestCase;
use Utopia\Cache\Adapter\None as NoCache;
use Utopia\Cache\Cache;
use Utopia\Database\Adapter\Memory;
use Utopia\Database\Database;
use Utopia\Database\Document;
use Utopia\Database\Helpers\Permission;
use Utopia\Database\Helpers\Role;
use Utopia\Database\Validator\Authorization;
use Utopia\Logger\Log;
use Utopia\Queue\Message;
use Utopia\Registry\Registry;
require_once __DIR__ . '/../../../../app/init.php';
/**
* Spy worker that records dispatch invocations instead of touching SMTP, the
* console alerts table, or external HTTP. Lets the worker tests assert
* routing, error handling, and alert persistence in isolation.
*/
class SpyNotifications extends Notifications
{
/** @var array<int, array{channel: string, address: string, signatureKey: ?string, payload: array<string, mixed>}> */
public array $dispatched = [];
/** @var array<string, \Throwable> */
public array $throwOn = [];
protected function dispatch(
array $recipient,
string $messageId,
array $payload,
Document $project,
Registry $register,
Database $database,
Log $log,
): ?string {
$channel = $recipient['channel'];
$this->dispatched[] = [
'channel' => $channel,
'address' => $recipient['address'],
'signatureKey' => $recipient['signatureKey'] ?? null,
'payload' => $payload,
];
if (isset($this->throwOn[$channel])) {
throw $this->throwOn[$channel];
}
// Mirror the real adapters' persistence contract so the action
// loop's branching (console/email persist internally; webhook
// persists in caller) is exercised end-to-end.
if ($messageId === '') {
return null;
}
if ($channel === NOTIFICATION_TYPE_CONSOLE || $channel === NOTIFICATION_TYPE_EMAIL) {
return $this->persistAlert($database, $messageId, $recipient, $payload);
}
return null;
}
}
class NotificationsTest extends TestCase
{
private Database $database;
private Authorization $authorization;
private Registry $registry;
private Document $project;
private Log $log;
protected function setUp(): void
{
$this->authorization = new Authorization();
$this->authorization->addRole(Role::any()->toString());
$this->database = new Database(new Memory(), new Cache(new NoCache()));
$this->database
->setAuthorization($this->authorization)
->setDatabase('notifTests')
->setNamespace('notif_' . \uniqid());
$this->database->create();
$this->database->createCollection(
'alerts',
[],
[],
[Permission::create(Role::any()), Permission::read(Role::any()), Permission::update(Role::any()), Permission::delete(Role::any())],
false,
);
$this->database->createAttribute('alerts', 'messageId', Database::VAR_STRING, 255, false);
$this->database->createAttribute('alerts', 'type', Database::VAR_STRING, 64, false, 'info');
$this->database->createAttribute('alerts', 'channel', Database::VAR_STRING, 64, true);
$this->database->createAttribute('alerts', 'userId', Database::VAR_STRING, 255, false);
$this->database->createAttribute('alerts', 'teamId', Database::VAR_STRING, 255, false);
$this->database->createAttribute('alerts', 'projectId', Database::VAR_STRING, 255, false);
$this->database->createAttribute('alerts', 'title', Database::VAR_STRING, 256, true);
$this->database->createAttribute('alerts', 'body', Database::VAR_STRING, 16384, true);
$this->database->createAttribute('alerts', 'read', Database::VAR_BOOLEAN, 0, false, false);
$this->registry = new Registry();
$this->project = new Document(['$id' => 'project-x']);
$this->log = new Log();
}
protected function tearDown(): void
{
$this->authorization->cleanRoles();
$this->authorization->addRole(Role::any()->toString());
}
private function buildMessage(array $payload): Message
{
return new Message([
'pid' => 'pid',
'queue' => 'v1-notifications',
'timestamp' => \time(),
'payload' => $payload,
]);
}
public function testDispatchesPerChannelToCorrectAdapter(): void
{
$worker = new SpyNotifications();
$payload = [
'project' => ['$id' => 'project-x'],
'recipients' => [
['address' => 'user@example.test', 'channel' => NOTIFICATION_TYPE_EMAIL],
['address' => 'user-1', 'channel' => NOTIFICATION_TYPE_CONSOLE],
['address' => 'https://hooks.example.test/in', 'channel' => NOTIFICATION_TYPE_WEBHOOK],
],
'subject' => 'Hi',
'body' => 'Body',
'deduplicationKey' => 'event-1',
];
$worker->action($this->buildMessage($payload), $this->project, $this->registry, $this->database, $this->log);
$this->assertCount(3, $worker->dispatched);
$channels = \array_map(static fn ($d) => $d['channel'], $worker->dispatched);
$this->assertSame([NOTIFICATION_TYPE_EMAIL, NOTIFICATION_TYPE_CONSOLE, NOTIFICATION_TYPE_WEBHOOK], $channels);
}
public function testPersistsOneAlertPerRecipientChannel(): void
{
$worker = new SpyNotifications();
$payload = [
'project' => ['$id' => 'project-x'],
'recipients' => [
['address' => 'user-1', 'channel' => NOTIFICATION_TYPE_CONSOLE],
['address' => 'user-2', 'channel' => NOTIFICATION_TYPE_CONSOLE],
],
'subject' => 'Heads up',
'body' => 'Read me',
'deduplicationKey' => 'evt-multi',
'permissions' => [Permission::read(Role::any())],
];
$worker->action($this->buildMessage($payload), $this->project, $this->registry, $this->database, $this->log);
$rows = $this->database->find('alerts');
$this->assertCount(2, $rows);
$userIds = \array_map(static fn (Document $row) => $row->getAttribute('userId'), $rows);
\sort($userIds);
$this->assertSame(['user-1', 'user-2'], $userIds);
foreach ($rows as $row) {
$this->assertSame(\md5('evt-multi'), $row->getAttribute('messageId'));
$this->assertSame('console', $row->getAttribute('channel'));
$this->assertSame('project-x', $row->getAttribute('projectId'));
$this->assertSame('Heads up', $row->getAttribute('title'));
}
}
public function testDedupHitShortCircuitsBeforeDispatch(): void
{
$worker = new SpyNotifications();
$payload = [
'project' => ['$id' => 'project-x'],
'recipients' => [['address' => 'user-1', 'channel' => NOTIFICATION_TYPE_CONSOLE]],
'subject' => 'Sub',
'body' => 'B',
'deduplicationKey' => 'dup-key',
];
// First run delivers and persists.
$worker->action($this->buildMessage($payload), $this->project, $this->registry, $this->database, $this->log);
$this->assertCount(1, $worker->dispatched);
// Manually insert a row with the dedup messageId so alreadyDelivered() returns true.
$messageId = \md5('dup-key');
$this->database->createDocument('alerts', new Document([
'$id' => $messageId,
'$permissions' => [Permission::read(Role::any())],
'messageId' => $messageId,
'channel' => 'console',
'title' => 'x',
'body' => 'y',
]));
$worker->dispatched = [];
$worker->action($this->buildMessage($payload), $this->project, $this->registry, $this->database, $this->log);
$this->assertCount(0, $worker->dispatched, 'second invocation must short-circuit on dedup hit');
}
public function testMissingRecipientsAndAddressThrows(): void
{
$worker = new SpyNotifications();
$this->expectException(\Exception::class);
$this->expectExceptionMessage('No recipients in payload');
$worker->action(
$this->buildMessage(['project' => ['$id' => 'project-x'], 'subject' => '', 'body' => '']),
$this->project,
$this->registry,
$this->database,
$this->log,
);
}
public function testFallbackToLegacyRecipient(): void
{
$worker = new SpyNotifications();
$payload = [
'project' => ['$id' => 'project-x'],
'recipient' => 'legacy@example.test',
'subject' => 'X',
'body' => 'Y',
];
$worker->action($this->buildMessage($payload), $this->project, $this->registry, $this->database, $this->log);
$this->assertCount(1, $worker->dispatched);
$this->assertSame('legacy@example.test', $worker->dispatched[0]['address']);
$this->assertSame(NOTIFICATION_TYPE_EMAIL, $worker->dispatched[0]['channel']);
}
public function testWebhookRecipientForwardsSignatureKey(): void
{
$worker = new SpyNotifications();
$payload = [
'project' => ['$id' => 'project-x'],
'recipients' => [
[
'address' => 'https://hooks.example.test/signed',
'channel' => NOTIFICATION_TYPE_WEBHOOK,
'signatureKey' => 'tenant-secret',
],
[
'address' => 'https://hooks.example.test/unsigned',
'channel' => NOTIFICATION_TYPE_WEBHOOK,
],
],
'subject' => 's',
'body' => 'b',
];
$worker->action($this->buildMessage($payload), $this->project, $this->registry, $this->database, $this->log);
$this->assertCount(2, $worker->dispatched);
$this->assertSame('tenant-secret', $worker->dispatched[0]['signatureKey']);
$this->assertNull($worker->dispatched[1]['signatureKey']);
}
public function testDispatchErrorTagsLogAndPropagates(): void
{
$worker = new SpyNotifications();
$worker->throwOn[NOTIFICATION_TYPE_WEBHOOK] = new \RuntimeException('boom');
$payload = [
'project' => ['$id' => 'project-x'],
'recipients' => [['address' => 'https://h.example.test', 'channel' => NOTIFICATION_TYPE_WEBHOOK]],
'subject' => 's',
'body' => 'b',
'deduplicationKey' => 'err-1',
];
try {
$worker->action($this->buildMessage($payload), $this->project, $this->registry, $this->database, $this->log);
$this->fail('expected exception to propagate');
} catch (\Throwable $error) {
$this->assertSame('boom', $error->getMessage());
}
$tags = $this->log->getTags();
$this->assertSame(NOTIFICATION_TYPE_WEBHOOK, $tags['channel'] ?? null);
$this->assertSame('boom', $tags['error'] ?? null);
$rows = $this->database->find('alerts');
$this->assertCount(0, $rows, 'failed dispatch must not persist alert');
}
}
@@ -0,0 +1,125 @@
<?php
namespace Tests\Unit\Utopia\Messaging\Adapter;
use Appwrite\Utopia\Messaging\Adapter\Console;
use Appwrite\Utopia\Messaging\Messages\Console as ConsoleMessage;
use PHPUnit\Framework\TestCase;
use Utopia\Cache\Adapter\None as NoCache;
use Utopia\Cache\Cache;
use Utopia\Database\Adapter\Memory;
use Utopia\Database\Database;
use Utopia\Database\Helpers\ID;
use Utopia\Database\Helpers\Permission;
use Utopia\Database\Helpers\Role;
use Utopia\Database\Validator\Authorization;
class ConsoleTest extends TestCase
{
private Database $database;
private Authorization $authorization;
protected function setUp(): void
{
$this->authorization = new Authorization();
$this->authorization->addRole(Role::any()->toString());
$this->database = new Database(new Memory(), new Cache(new NoCache()));
$this->database
->setAuthorization($this->authorization)
->setDatabase('alertsTests')
->setNamespace('alerts_' . \uniqid());
$this->database->create();
$this->database->createCollection('alerts', [], [], [Permission::create(Role::any()), Permission::read(Role::any())], false);
$this->database->createAttribute('alerts', 'messageId', Database::VAR_STRING, 255, false);
$this->database->createAttribute('alerts', 'type', Database::VAR_STRING, 64, false, 'info');
$this->database->createAttribute('alerts', 'channel', Database::VAR_STRING, 64, true);
$this->database->createAttribute('alerts', 'userId', Database::VAR_STRING, 255, false);
$this->database->createAttribute('alerts', 'teamId', Database::VAR_STRING, 255, false);
$this->database->createAttribute('alerts', 'projectId', Database::VAR_STRING, 255, false);
$this->database->createAttribute('alerts', 'title', Database::VAR_STRING, 256, true);
$this->database->createAttribute('alerts', 'body', Database::VAR_STRING, 16384, true);
}
protected function tearDown(): void
{
$this->authorization->cleanRoles();
$this->authorization->addRole(Role::any()->toString());
}
public function testWritesAlertWithCorrectSchema(): void
{
$message = new ConsoleMessage(
recipients: [['userId' => 'user-1']],
title: 'Hello',
body: 'World',
type: 'info',
messageId: ID::custom('msg-aaa'),
projectId: 'project-1',
);
$adapter = new Console($this->database);
$result = $adapter->send($message);
$this->assertSame(1, $result['deliveredTo']);
$stored = $this->database->getDocument('alerts', 'msg-aaa');
$this->assertFalse($stored->isEmpty());
$this->assertSame('msg-aaa', $stored->getAttribute('messageId'));
$this->assertSame('console', $stored->getAttribute('channel'));
$this->assertSame('user-1', $stored->getAttribute('userId'));
$this->assertSame('project-1', $stored->getAttribute('projectId'));
$this->assertSame('Hello', $stored->getAttribute('title'));
$this->assertSame('World', $stored->getAttribute('body'));
$this->assertSame('info', $stored->getAttribute('type'));
}
public function testUserPermissionsScopedToRecipient(): void
{
$message = new ConsoleMessage(
recipients: [['userId' => 'user-2']],
title: 'Title',
body: 'Body',
messageId: ID::custom('msg-perms-user'),
);
(new Console($this->database))->send($message);
$stored = $this->database->getDocument('alerts', 'msg-perms-user');
$permissions = $stored->getPermissions();
$this->assertContains(Permission::read(Role::user('user-2')), $permissions);
$this->assertContains(Permission::update(Role::user('user-2')), $permissions);
$this->assertContains(Permission::delete(Role::user('user-2')), $permissions);
}
public function testTeamRecipientGrantsTeamReadAndOwnerWrite(): void
{
$message = new ConsoleMessage(
recipients: [['teamId' => 'team-9']],
title: 'Heads up',
body: '...',
messageId: ID::custom('msg-team'),
);
(new Console($this->database))->send($message);
$stored = $this->database->getDocument('alerts', 'msg-team');
$permissions = $stored->getPermissions();
$this->assertContains(Permission::read(Role::team('team-9')), $permissions);
$this->assertContains(Permission::update(Role::team('team-9', 'owner')), $permissions);
$this->assertContains(Permission::delete(Role::team('team-9', 'owner')), $permissions);
}
public function testRejectsForeignMessageType(): void
{
$adapter = new Console($this->database);
$this->expectException(\Exception::class);
$this->expectExceptionMessage('Invalid message type.');
// ConsoleMessage extends nothing — pass an unrelated Message implementation
$adapter->send(new \Appwrite\Utopia\Messaging\Messages\Webhook(urls: ['https://example.test'], payload: []));
}
}
@@ -0,0 +1,195 @@
<?php
namespace Tests\Unit\Utopia\Messaging\Adapter;
use Appwrite\Utopia\Messaging\Adapter\Webhook;
use Appwrite\Utopia\Messaging\Messages\Webhook as WebhookMessage;
use PHPUnit\Framework\TestCase;
/**
* Test double that captures the curl request the adapter would issue and
* returns a scripted response, so we exercise the real signing/header logic
* without touching the network.
*/
class CapturingWebhook extends Webhook
{
/**
* @var array<int, array{method: string, url: string, headers: array<int, string>, body: string, timeout: int}>
*/
public array $captured = [];
/** @var array{statusCode: int, response: string|null, error: string|null} */
public array $response = ['statusCode' => 200, 'response' => 'OK', 'error' => null];
protected function dispatch(string $method, string $url, array $headers, string $body, int $timeout): array
{
$this->captured[] = [
'method' => $method,
'url' => $url,
'headers' => $headers,
'body' => $body,
'timeout' => $timeout,
];
return $this->response;
}
}
class WebhookTest extends TestCase
{
public function testPostsExpectedBodyShape(): void
{
$adapter = new CapturingWebhook();
$payload = [
'subject' => 'Hello',
'body' => 'World',
'recipient' => 'ops@example.test',
'metadata' => ['foo' => 'bar'],
];
$message = new WebhookMessage(
urls: ['https://hooks.example.test/notify'],
payload: $payload,
);
$result = $adapter->send($message);
$this->assertSame(1, $result['deliveredTo']);
$this->assertCount(1, $adapter->captured);
$request = $adapter->captured[0];
$this->assertSame('POST', $request['method']);
$this->assertSame('https://hooks.example.test/notify', $request['url']);
$sent = \json_decode($request['body'], true);
$this->assertSame($payload, $sent);
$headerLine = \implode("\n", $request['headers']);
$this->assertStringContainsString('Content-Type: application/json', $headerLine);
$this->assertStringContainsString('X-Appwrite-Webhook-Timestamp:', $headerLine);
}
public function testSigningSecretProducesHmacSha256Signature(): void
{
$adapter = new CapturingWebhook();
$payload = ['subject' => 'Signed', 'body' => 'B'];
$secret = 'super-secret';
$message = new WebhookMessage(
urls: ['https://hooks.example.test/signed'],
payload: $payload,
signingSecret: $secret,
);
$adapter->send($message);
$headers = $adapter->captured[0]['headers'];
$body = $adapter->captured[0]['body'];
$timestamp = null;
$signature = null;
foreach ($headers as $header) {
if (\str_starts_with($header, 'X-Appwrite-Webhook-Timestamp: ')) {
$timestamp = \substr($header, \strlen('X-Appwrite-Webhook-Timestamp: '));
} elseif (\str_starts_with($header, 'X-Appwrite-Webhook-Signature: ')) {
$signature = \substr($header, \strlen('X-Appwrite-Webhook-Signature: '));
}
}
$this->assertNotNull($timestamp, 'timestamp header must be present');
$this->assertNotNull($signature, 'signature header must be present when secret is set');
$this->assertStringStartsWith('sha256=', $signature);
$expected = 'sha256=' . \hash_hmac('sha256', $timestamp . '.' . $body, $secret);
$this->assertSame($expected, $signature);
}
public function testNoSecretLeavesPayloadUnsigned(): void
{
$adapter = new CapturingWebhook();
$message = new WebhookMessage(
urls: ['https://hooks.example.test/unsigned'],
payload: ['x' => 1],
signingSecret: null,
);
$adapter->send($message);
$headerLine = \implode("\n", $adapter->captured[0]['headers']);
$this->assertStringNotContainsString('X-Appwrite-Webhook-Signature', $headerLine);
}
public function testEmptySecretIsTreatedAsUnsigned(): void
{
$adapter = new CapturingWebhook();
$message = new WebhookMessage(
urls: ['https://hooks.example.test/empty-secret'],
payload: ['x' => 1],
signingSecret: '',
);
$adapter->send($message);
$headerLine = \implode("\n", $adapter->captured[0]['headers']);
$this->assertStringNotContainsString('X-Appwrite-Webhook-Signature', $headerLine);
}
public function testTwoXxIsSuccess(): void
{
$adapter = new CapturingWebhook();
$adapter->response = ['statusCode' => 204, 'response' => '', 'error' => null];
$message = new WebhookMessage(urls: ['https://hooks.example.test/ok'], payload: []);
$result = $adapter->send($message);
$this->assertSame(1, $result['deliveredTo']);
}
public function testNonTwoXxSurfacesError(): void
{
$adapter = new CapturingWebhook();
$adapter->response = ['statusCode' => 503, 'response' => 'Server', 'error' => null];
$message = new WebhookMessage(urls: ['https://hooks.example.test/fail'], payload: []);
$result = $adapter->send($message);
$this->assertSame(0, $result['deliveredTo']);
$error = $result['results'][0]['error'] ?? null;
$this->assertSame('HTTP 503', $error);
}
public function testCurlErrorSurfacesAsResultError(): void
{
$adapter = new CapturingWebhook();
$adapter->response = ['statusCode' => 0, 'response' => null, 'error' => 'connection refused'];
$message = new WebhookMessage(urls: ['https://hooks.example.test/down'], payload: []);
$result = $adapter->send($message);
$this->assertSame(0, $result['deliveredTo']);
$this->assertSame('connection refused', $result['results'][0]['error']);
}
public function testCustomHeadersForwarded(): void
{
$adapter = new CapturingWebhook();
$message = new WebhookMessage(
urls: ['https://hooks.example.test/with-headers'],
payload: [],
headers: ['X-Custom' => 'value'],
);
$adapter->send($message);
$headerLine = \implode("\n", $adapter->captured[0]['headers']);
$this->assertStringContainsString('X-Custom: value', $headerLine);
}
public function testRejectsForeignMessageType(): void
{
$adapter = new CapturingWebhook();
$this->expectException(\Exception::class);
$this->expectExceptionMessage('Invalid message type.');
$adapter->send(new \Appwrite\Utopia\Messaging\Messages\Console(recipients: [['userId' => 'u']], title: 't', body: 'b'));
}
}