You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@james.apache.org by rc...@apache.org on 2024/01/29 09:52:56 UTC
(james-project) branch postgresql updated: JAMES-2586 Implement Postgres Push subscription
This is an automated email from the ASF dual-hosted git repository.
rcordier pushed a commit to branch postgresql
in repository https://gitbox.apache.org/repos/asf/james-project.git
The following commit(s) were added to refs/heads/postgresql by this push:
new 553868f628 JAMES-2586 Implement Postgres Push subscription
553868f628 is described below
commit 553868f6289e410a205a9d6309fc7eddd5f9a073
Author: Tung Tran <vt...@linagora.com>
AuthorDate: Fri Jan 26 13:44:54 2024 +0700
JAMES-2586 Implement Postgres Push subscription
---
.../james/backends/postgres/PostgresCommons.java | 8 +-
.../org/apache/james/PostgresJamesServerMain.java | 2 -
.../java/org/apache/james/PostgresJmapModule.java | 4 +
.../postgres/PostgresDataJMapAggregateModule.java | 4 +-
.../PostgresPushSubscriptionDAO.java | 171 +++++++++++++++++++++
.../PostgresPushSubscriptionModule.java | 80 ++++++++++
.../PostgresPushSubscriptionRepository.java | 140 +++++++++++++++++
.../PostgresPushSubscriptionRepositoryTest.java | 62 ++++++++
8 files changed, 467 insertions(+), 4 deletions(-)
diff --git a/backends-common/postgres/src/main/java/org/apache/james/backends/postgres/PostgresCommons.java b/backends-common/postgres/src/main/java/org/apache/james/backends/postgres/PostgresCommons.java
index 88d36936c8..d465740e40 100644
--- a/backends-common/postgres/src/main/java/org/apache/james/backends/postgres/PostgresCommons.java
+++ b/backends-common/postgres/src/main/java/org/apache/james/backends/postgres/PostgresCommons.java
@@ -21,7 +21,9 @@ package org.apache.james.backends.postgres;
import java.time.Instant;
import java.time.LocalDateTime;
+import java.time.ZoneId;
import java.time.ZoneOffset;
+import java.time.ZonedDateTime;
import java.util.Date;
import java.util.Optional;
import java.util.function.Function;
@@ -47,7 +49,7 @@ public class PostgresCommons {
DataType<LocalDateTime> TIMESTAMP = SQLDataType.LOCALDATETIME(6);
// text[]
- DataType<String[]> STRING_ARRAY = SQLDataType.CLOB.getArrayDataType();
+ DataType<String[]> STRING_ARRAY = SQLDataType.VARCHAR.getArrayDataType();
}
@@ -68,6 +70,10 @@ public class PostgresCommons {
.map(Date::from)
.orElse(null);
+ public static final Function<LocalDateTime, ZonedDateTime> LOCAL_DATE_TIME_ZONED_DATE_TIME_FUNCTION = localDateTime -> Optional.ofNullable(localDateTime)
+ .map(value -> value.atZone(ZoneId.of("UTC")))
+ .orElse(null);
+
public static final Function<LocalDateTime, Instant> LOCAL_DATE_TIME_INSTANT_FUNCTION = localDateTime -> Optional.ofNullable(localDateTime)
.map(value -> value.toInstant(ZoneOffset.UTC))
.orElse(null);
diff --git a/server/apps/postgres-app/src/main/java/org/apache/james/PostgresJamesServerMain.java b/server/apps/postgres-app/src/main/java/org/apache/james/PostgresJamesServerMain.java
index 0a4a2c8590..367148dcbe 100644
--- a/server/apps/postgres-app/src/main/java/org/apache/james/PostgresJamesServerMain.java
+++ b/server/apps/postgres-app/src/main/java/org/apache/james/PostgresJamesServerMain.java
@@ -23,7 +23,6 @@ import java.util.List;
import org.apache.james.data.UsersRepositoryModuleChooser;
import org.apache.james.jmap.draft.JMAPListenerModule;
-import org.apache.james.jmap.memory.pushsubscription.MemoryPushSubscriptionModule;
import org.apache.james.modules.BlobExportMechanismModule;
import org.apache.james.modules.MailboxModule;
import org.apache.james.modules.MailetProcessingModule;
@@ -120,7 +119,6 @@ public class PostgresJamesServerMain implements JamesServerMain {
public static final Module JMAP = Modules.combine(
new PostgresJmapModule(),
new PostgresDataJmapModule(),
- new MemoryPushSubscriptionModule(),
new JmapEventBusModule(),
new JMAPServerModule(),
new JmapTasksModule());
diff --git a/server/apps/postgres-app/src/main/java/org/apache/james/PostgresJmapModule.java b/server/apps/postgres-app/src/main/java/org/apache/james/PostgresJmapModule.java
index 1dca0ac9be..018982f312 100644
--- a/server/apps/postgres-app/src/main/java/org/apache/james/PostgresJmapModule.java
+++ b/server/apps/postgres-app/src/main/java/org/apache/james/PostgresJmapModule.java
@@ -24,12 +24,14 @@ import org.apache.james.jmap.api.change.EmailChangeRepository;
import org.apache.james.jmap.api.change.Limit;
import org.apache.james.jmap.api.change.MailboxChangeRepository;
import org.apache.james.jmap.api.change.State;
+import org.apache.james.jmap.api.pushsubscription.PushSubscriptionRepository;
import org.apache.james.jmap.api.upload.UploadUsageRepository;
import org.apache.james.jmap.memory.change.MemoryEmailChangeRepository;
import org.apache.james.jmap.memory.change.MemoryMailboxChangeRepository;
import org.apache.james.jmap.postgres.PostgresDataJMapAggregateModule;
import org.apache.james.jmap.postgres.change.PostgresEmailChangeModule;
import org.apache.james.jmap.postgres.change.PostgresEmailChangeRepository;
+import org.apache.james.jmap.postgres.pushsubscription.PostgresPushSubscriptionRepository;
import org.apache.james.jmap.postgres.upload.PostgresUploadUsageRepository;
import org.apache.james.mailbox.AttachmentManager;
import org.apache.james.mailbox.MessageIdManager;
@@ -84,5 +86,7 @@ public class PostgresJmapModule extends AbstractModule {
bind(StoreRightManager.class).in(Scopes.SINGLETON);
bind(State.Factory.class).toInstance(State.Factory.DEFAULT);
+
+ bind(PushSubscriptionRepository.class).to(PostgresPushSubscriptionRepository.class);
}
}
diff --git a/server/data/data-jmap-postgres/src/main/java/org/apache/james/jmap/postgres/PostgresDataJMapAggregateModule.java b/server/data/data-jmap-postgres/src/main/java/org/apache/james/jmap/postgres/PostgresDataJMapAggregateModule.java
index 28459dbf97..592f8fb2b2 100644
--- a/server/data/data-jmap-postgres/src/main/java/org/apache/james/jmap/postgres/PostgresDataJMapAggregateModule.java
+++ b/server/data/data-jmap-postgres/src/main/java/org/apache/james/jmap/postgres/PostgresDataJMapAggregateModule.java
@@ -22,6 +22,7 @@ package org.apache.james.jmap.postgres;
import org.apache.james.backends.postgres.PostgresModule;
import org.apache.james.jmap.postgres.change.PostgresEmailChangeModule;
import org.apache.james.jmap.postgres.projections.PostgresMessageFastViewProjectionModule;
+import org.apache.james.jmap.postgres.pushsubscription.PostgresPushSubscriptionModule;
import org.apache.james.jmap.postgres.upload.PostgresUploadModule;
public interface PostgresDataJMapAggregateModule {
@@ -29,5 +30,6 @@ public interface PostgresDataJMapAggregateModule {
PostgresModule MODULE = PostgresModule.aggregateModules(
PostgresUploadModule.MODULE,
PostgresMessageFastViewProjectionModule.MODULE,
- PostgresEmailChangeModule.MODULE);
+ PostgresEmailChangeModule.MODULE,
+ PostgresPushSubscriptionModule.MODULE);
}
diff --git a/server/data/data-jmap-postgres/src/main/java/org/apache/james/jmap/postgres/pushsubscription/PostgresPushSubscriptionDAO.java b/server/data/data-jmap-postgres/src/main/java/org/apache/james/jmap/postgres/pushsubscription/PostgresPushSubscriptionDAO.java
new file mode 100644
index 0000000000..69f0abf41c
--- /dev/null
+++ b/server/data/data-jmap-postgres/src/main/java/org/apache/james/jmap/postgres/pushsubscription/PostgresPushSubscriptionDAO.java
@@ -0,0 +1,171 @@
+/****************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one *
+ * or more contributor license agreements. See the NOTICE file *
+ * distributed with this work for additional information *
+ * regarding copyright ownership. The ASF licenses this file *
+ * to you under the Apache License, Version 2.0 (the *
+ * "License"); you may not use this file except in compliance *
+ * with the License. You may obtain a copy of the License at *
+ * *
+ * http://www.apache.org/licenses/LICENSE-2.0 *
+ * *
+ * Unless required by applicable law or agreed to in writing, *
+ * software distributed under the License is distributed on an *
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY *
+ * KIND, either express or implied. See the License for the *
+ * specific language governing permissions and limitations *
+ * under the License. *
+ ****************************************************************/
+
+package org.apache.james.jmap.postgres.pushsubscription;
+
+import static org.apache.james.backends.postgres.PostgresCommons.IN_CLAUSE_MAX_SIZE;
+import static org.apache.james.backends.postgres.PostgresCommons.LOCAL_DATE_TIME_ZONED_DATE_TIME_FUNCTION;
+
+import java.time.LocalDateTime;
+import java.time.ZonedDateTime;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Optional;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import org.apache.james.backends.postgres.utils.PostgresExecutor;
+import org.apache.james.core.Username;
+import org.apache.james.jmap.api.change.TypeStateFactory;
+import org.apache.james.jmap.api.model.PushSubscription;
+import org.apache.james.jmap.api.model.PushSubscriptionExpiredTime;
+import org.apache.james.jmap.api.model.PushSubscriptionId;
+import org.apache.james.jmap.api.model.PushSubscriptionKeys;
+import org.apache.james.jmap.api.model.PushSubscriptionServerURL;
+import org.apache.james.jmap.api.model.TypeName;
+import org.apache.james.jmap.postgres.pushsubscription.PostgresPushSubscriptionModule.PushSubscriptionTable;
+import org.jooq.Record;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Iterables;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import scala.jdk.javaapi.CollectionConverters;
+import scala.jdk.javaapi.OptionConverters;
+
+public class PostgresPushSubscriptionDAO {
+ private final PostgresExecutor postgresExecutor;
+ private final TypeStateFactory typeStateFactory;
+
+ public PostgresPushSubscriptionDAO(PostgresExecutor postgresExecutor, TypeStateFactory typeStateFactory) {
+ this.postgresExecutor = postgresExecutor;
+ this.typeStateFactory = typeStateFactory;
+ }
+
+ public Mono<Void> save(Username username, PushSubscription pushSubscription) {
+ return postgresExecutor.executeVoid(dslContext -> Mono.from(dslContext.insertInto(PushSubscriptionTable.TABLE_NAME)
+ .set(PushSubscriptionTable.USER, username.asString())
+ .set(PushSubscriptionTable.DEVICE_CLIENT_ID, pushSubscription.deviceClientId())
+ .set(PushSubscriptionTable.ID, pushSubscription.id().value())
+ .set(PushSubscriptionTable.EXPIRES, pushSubscription.expires().value().toLocalDateTime())
+ .set(PushSubscriptionTable.TYPES, CollectionConverters.asJava(pushSubscription.types())
+ .stream().map(TypeName::asString).toArray(String[]::new))
+ .set(PushSubscriptionTable.URL, pushSubscription.url().value().toString())
+ .set(PushSubscriptionTable.VERIFICATION_CODE, pushSubscription.verificationCode())
+ .set(PushSubscriptionTable.VALIDATED, pushSubscription.validated())
+ .set(PushSubscriptionTable.ENCRYPT_PUBLIC_KEY, OptionConverters.toJava(pushSubscription.keys().map(PushSubscriptionKeys::p256dh)).orElse(null))
+ .set(PushSubscriptionTable.ENCRYPT_AUTH_SECRET, OptionConverters.toJava(pushSubscription.keys().map(PushSubscriptionKeys::auth)).orElse(null))));
+ }
+
+ public Flux<PushSubscription> listByUsername(Username username) {
+ return postgresExecutor.executeRows(dslContext -> Flux.from(dslContext.selectFrom(PushSubscriptionTable.TABLE_NAME)
+ .where(PushSubscriptionTable.USER.eq(username.asString()))))
+ .map(this::recordAsPushSubscription);
+ }
+
+ public Flux<PushSubscription> getByUsernameAndIds(Username username, Collection<PushSubscriptionId> ids) {
+ Function<Collection<PushSubscriptionId>, Flux<PushSubscription>> queryPublisherFunction = idsMatching -> postgresExecutor.executeRows(dslContext -> Flux.from(dslContext.selectFrom(PushSubscriptionTable.TABLE_NAME)
+ .where(PushSubscriptionTable.USER.eq(username.asString()))
+ .and(PushSubscriptionTable.ID.in(idsMatching.stream().map(PushSubscriptionId::value).collect(Collectors.toList())))))
+ .map(this::recordAsPushSubscription);
+
+ if (ids.size() <= IN_CLAUSE_MAX_SIZE) {
+ return queryPublisherFunction.apply(ids);
+ } else {
+ return Flux.fromIterable(Iterables.partition(ids, IN_CLAUSE_MAX_SIZE))
+ .flatMap(queryPublisherFunction);
+ }
+ }
+
+ public Mono<Void> deleteByUsername(Username username) {
+ return postgresExecutor.executeVoid(dslContext -> Mono.from(dslContext.deleteFrom(PushSubscriptionTable.TABLE_NAME)
+ .where(PushSubscriptionTable.USER.eq(username.asString()))));
+ }
+
+ public Mono<Void> deleteByUsernameAndId(Username username, PushSubscriptionId id) {
+ return postgresExecutor.executeVoid(dslContext -> Mono.from(dslContext.deleteFrom(PushSubscriptionTable.TABLE_NAME)
+ .where(PushSubscriptionTable.USER.eq(username.asString()))
+ .and(PushSubscriptionTable.ID.eq(id.value()))));
+ }
+
+ public Mono<Set<TypeName>> updateType(Username username, PushSubscriptionId id, Set<TypeName> newTypes) {
+ Preconditions.checkNotNull(newTypes, "newTypes should not be null");
+ return postgresExecutor.executeRow(dslContext -> Mono.from(dslContext.update(PushSubscriptionTable.TABLE_NAME)
+ .set(PushSubscriptionTable.TYPES, newTypes.stream().map(TypeName::asString).toArray(String[]::new))
+ .where(PushSubscriptionTable.USER.eq(username.asString()))
+ .and(PushSubscriptionTable.ID.eq(id.value()))
+ .returning(PushSubscriptionTable.TYPES)))
+ .map(this::extractTypes);
+ }
+
+ public Mono<Boolean> updateValidated(Username username, PushSubscriptionId id, boolean validated) {
+ return postgresExecutor.executeRow(dslContext -> Mono.from(dslContext.update(PushSubscriptionTable.TABLE_NAME)
+ .set(PushSubscriptionTable.VALIDATED, validated)
+ .where(PushSubscriptionTable.USER.eq(username.asString()))
+ .and(PushSubscriptionTable.ID.eq(id.value()))
+ .returning(PushSubscriptionTable.VALIDATED)))
+ .map(record -> record.get(PushSubscriptionTable.VALIDATED));
+ }
+
+ public Mono<ZonedDateTime> updateExpireTime(Username username, PushSubscriptionId id, ZonedDateTime newExpire) {
+ Preconditions.checkNotNull(newExpire, "newExpire should not be null");
+ return postgresExecutor.executeRow(dslContext -> Mono.from(dslContext.update(PushSubscriptionTable.TABLE_NAME)
+ .set(PushSubscriptionTable.EXPIRES, newExpire.toLocalDateTime())
+ .where(PushSubscriptionTable.USER.eq(username.asString()))
+ .and(PushSubscriptionTable.ID.eq(id.value()))
+ .returning(PushSubscriptionTable.EXPIRES)))
+ .map(record -> LOCAL_DATE_TIME_ZONED_DATE_TIME_FUNCTION.apply(record.get(PushSubscriptionTable.EXPIRES)));
+ }
+
+ public Mono<Boolean> existDeviceClientId(Username username, String deviceClientId) {
+ return postgresExecutor.executeRow(dslContext -> Mono.from(dslContext.select(PushSubscriptionTable.DEVICE_CLIENT_ID)
+ .from(PushSubscriptionTable.TABLE_NAME)
+ .where(PushSubscriptionTable.USER.eq(username.asString()))
+ .and(PushSubscriptionTable.DEVICE_CLIENT_ID.eq(deviceClientId))
+ .limit(1)))
+ .hasElement();
+ }
+
+ private PushSubscription recordAsPushSubscription(Record record) {
+ try {
+ return new PushSubscription(new PushSubscriptionId(record.get(PushSubscriptionTable.ID)),
+ record.get(PushSubscriptionTable.DEVICE_CLIENT_ID),
+ PushSubscriptionServerURL.from(record.get(PushSubscriptionTable.URL)).get(),
+ scala.jdk.javaapi.OptionConverters.toScala(Optional.ofNullable(record.get(PushSubscriptionTable.ENCRYPT_PUBLIC_KEY))
+ .flatMap(key -> Optional.ofNullable(record.get(PushSubscriptionTable.ENCRYPT_AUTH_SECRET))
+ .map(secret -> new PushSubscriptionKeys(key, secret)))),
+ record.get(PushSubscriptionTable.VERIFICATION_CODE),
+ record.get(PushSubscriptionTable.VALIDATED),
+ Optional.ofNullable(record.get(PushSubscriptionTable.EXPIRES, LocalDateTime.class))
+ .map(LOCAL_DATE_TIME_ZONED_DATE_TIME_FUNCTION)
+ .map(PushSubscriptionExpiredTime::new).get(),
+ CollectionConverters.asScala(extractTypes(record)).toSeq());
+ } catch (Exception e) {
+ throw new RuntimeException("Error while parsing PushSubscription from database", e);
+ }
+ }
+
+ private Set<TypeName> extractTypes(Record record) {
+ return Arrays.stream(record.get(PushSubscriptionTable.TYPES))
+ .map(string -> typeStateFactory.parse(string).right().get())
+ .collect(Collectors.toSet());
+ }
+}
\ No newline at end of file
diff --git a/server/data/data-jmap-postgres/src/main/java/org/apache/james/jmap/postgres/pushsubscription/PostgresPushSubscriptionModule.java b/server/data/data-jmap-postgres/src/main/java/org/apache/james/jmap/postgres/pushsubscription/PostgresPushSubscriptionModule.java
new file mode 100644
index 0000000000..eceda1c2b1
--- /dev/null
+++ b/server/data/data-jmap-postgres/src/main/java/org/apache/james/jmap/postgres/pushsubscription/PostgresPushSubscriptionModule.java
@@ -0,0 +1,80 @@
+/****************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one *
+ * or more contributor license agreements. See the NOTICE file *
+ * distributed with this work for additional information *
+ * regarding copyright ownership. The ASF licenses this file *
+ * to you under the Apache License, Version 2.0 (the *
+ * "License"); you may not use this file except in compliance *
+ * with the License. You may obtain a copy of the License at *
+ * *
+ * http://www.apache.org/licenses/LICENSE-2.0 *
+ * *
+ * Unless required by applicable law or agreed to in writing, *
+ * software distributed under the License is distributed on an *
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY *
+ * KIND, either express or implied. See the License for the *
+ * specific language governing permissions and limitations *
+ * under the License. *
+ ****************************************************************/
+
+package org.apache.james.jmap.postgres.pushsubscription;
+
+import java.time.LocalDateTime;
+import java.util.UUID;
+
+import org.apache.james.backends.postgres.PostgresCommons;
+import org.apache.james.backends.postgres.PostgresIndex;
+import org.apache.james.backends.postgres.PostgresModule;
+import org.apache.james.backends.postgres.PostgresTable;
+import org.jooq.Field;
+import org.jooq.Record;
+import org.jooq.Table;
+import org.jooq.impl.DSL;
+import org.jooq.impl.SQLDataType;
+
+public interface PostgresPushSubscriptionModule {
+
+ interface PushSubscriptionTable {
+ Table<Record> TABLE_NAME = DSL.table("push_subscription");
+ Field<String> USER = DSL.field("username", SQLDataType.VARCHAR.notNull());
+ Field<String> DEVICE_CLIENT_ID = DSL.field("device_client_id", SQLDataType.VARCHAR.notNull());
+
+ Field<UUID> ID = DSL.field("id", SQLDataType.UUID.notNull());
+ Field<LocalDateTime> EXPIRES = DSL.field("expires", PostgresCommons.DataTypes.TIMESTAMP);
+ Field<String[]> TYPES = DSL.field("types", PostgresCommons.DataTypes.STRING_ARRAY.notNull());
+
+ Field<String> URL = DSL.field("url", SQLDataType.VARCHAR.notNull());
+ Field<String> VERIFICATION_CODE = DSL.field("verification_code", SQLDataType.VARCHAR);
+ Field<String> ENCRYPT_PUBLIC_KEY = DSL.field("encrypt_public_key", SQLDataType.VARCHAR);
+ Field<String> ENCRYPT_AUTH_SECRET = DSL.field("encrypt_auth_secret", SQLDataType.VARCHAR);
+ Field<Boolean> VALIDATED = DSL.field("validated", SQLDataType.BOOLEAN.notNull());
+
+ PostgresTable TABLE = PostgresTable.name(TABLE_NAME.getName())
+ .createTableStep(((dsl, tableName) -> dsl.createTableIfNotExists(tableName)
+ .column(USER)
+ .column(DEVICE_CLIENT_ID)
+ .column(ID)
+ .column(EXPIRES)
+ .column(TYPES)
+ .column(URL)
+ .column(VERIFICATION_CODE)
+ .column(ENCRYPT_PUBLIC_KEY)
+ .column(ENCRYPT_AUTH_SECRET)
+ .column(VALIDATED)
+ .primaryKey(USER, DEVICE_CLIENT_ID)))
+ .supportsRowLevelSecurity()
+ .build();
+
+ PostgresIndex USERNAME_INDEX = PostgresIndex.name("push_subscription_username_index")
+ .createIndexStep((dslContext, indexName) -> dslContext.createIndexIfNotExists(indexName)
+ .on(TABLE_NAME, USER));
+ PostgresIndex USERNAME_ID_INDEX = PostgresIndex.name("push_subscription_username_id_index")
+ .createIndexStep((dslContext, indexName) -> dslContext.createIndexIfNotExists(indexName)
+ .on(TABLE_NAME, USER, ID));
+ }
+
+ PostgresModule MODULE = PostgresModule.builder()
+ .addTable(PushSubscriptionTable.TABLE)
+ .addIndex(PushSubscriptionTable.USERNAME_INDEX, PushSubscriptionTable.USERNAME_ID_INDEX)
+ .build();
+}
diff --git a/server/data/data-jmap-postgres/src/main/java/org/apache/james/jmap/postgres/pushsubscription/PostgresPushSubscriptionRepository.java b/server/data/data-jmap-postgres/src/main/java/org/apache/james/jmap/postgres/pushsubscription/PostgresPushSubscriptionRepository.java
new file mode 100644
index 0000000000..c2370e43ad
--- /dev/null
+++ b/server/data/data-jmap-postgres/src/main/java/org/apache/james/jmap/postgres/pushsubscription/PostgresPushSubscriptionRepository.java
@@ -0,0 +1,140 @@
+/****************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one *
+ * or more contributor license agreements. See the NOTICE file *
+ * distributed with this work for additional information *
+ * regarding copyright ownership. The ASF licenses this file *
+ * to you under the Apache License, Version 2.0 (the *
+ * "License"); you may not use this file except in compliance *
+ * with the License. You may obtain a copy of the License at *
+ * *
+ * http://www.apache.org/licenses/LICENSE-2.0 *
+ * *
+ * Unless required by applicable law or agreed to in writing, *
+ * software distributed under the License is distributed on an *
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY *
+ * KIND, either express or implied. See the License for the *
+ * specific language governing permissions and limitations *
+ * under the License. *
+ ****************************************************************/
+
+package org.apache.james.jmap.postgres.pushsubscription;
+
+import static org.apache.james.jmap.api.pushsubscription.PushSubscriptionHelpers.evaluateExpiresTime;
+import static org.apache.james.jmap.api.pushsubscription.PushSubscriptionHelpers.isInThePast;
+import static org.apache.james.jmap.api.pushsubscription.PushSubscriptionHelpers.isInvalidPushSubscriptionKey;
+
+import java.time.Clock;
+import java.time.ZonedDateTime;
+import java.util.Optional;
+import java.util.Set;
+
+import javax.inject.Inject;
+import javax.inject.Singleton;
+
+import org.apache.james.backends.postgres.utils.PostgresExecutor;
+import org.apache.james.core.Username;
+import org.apache.james.jmap.api.change.TypeStateFactory;
+import org.apache.james.jmap.api.model.DeviceClientIdInvalidException;
+import org.apache.james.jmap.api.model.ExpireTimeInvalidException;
+import org.apache.james.jmap.api.model.InvalidPushSubscriptionKeys;
+import org.apache.james.jmap.api.model.PushSubscription;
+import org.apache.james.jmap.api.model.PushSubscriptionCreationRequest;
+import org.apache.james.jmap.api.model.PushSubscriptionExpiredTime;
+import org.apache.james.jmap.api.model.PushSubscriptionId;
+import org.apache.james.jmap.api.model.PushSubscriptionNotFoundException;
+import org.apache.james.jmap.api.model.TypeName;
+import org.apache.james.jmap.api.pushsubscription.PushSubscriptionRepository;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import scala.jdk.javaapi.OptionConverters;
+
+public class PostgresPushSubscriptionRepository implements PushSubscriptionRepository {
+ private final Clock clock;
+ private final TypeStateFactory typeStateFactory;
+ private final PostgresExecutor.Factory executorFactory;
+
+ @Inject
+ @Singleton
+ public PostgresPushSubscriptionRepository(Clock clock, TypeStateFactory typeStateFactory, PostgresExecutor.Factory executorFactory) {
+ this.clock = clock;
+ this.typeStateFactory = typeStateFactory;
+ this.executorFactory = executorFactory;
+ }
+
+ @Override
+ public Mono<PushSubscription> save(Username username, PushSubscriptionCreationRequest request) {
+ PushSubscription pushSubscription = PushSubscription.from(request,
+ evaluateExpiresTime(OptionConverters.toJava(request.expires().map(PushSubscriptionExpiredTime::value)), clock));
+
+ PostgresPushSubscriptionDAO pushSubscriptionDAO = getDAO(username);
+ return pushSubscriptionDAO.existDeviceClientId(username, request.deviceClientId())
+ .handle((isDuplicated, sink) -> {
+ if (isInThePast(request.expires(), clock)) {
+ sink.error(new ExpireTimeInvalidException(request.expires().get().value(), "expires must be greater than now"));
+ return;
+ }
+ if (isDuplicated) {
+ sink.error(new DeviceClientIdInvalidException(request.deviceClientId(), "deviceClientId must be unique"));
+ return;
+ }
+ if (isInvalidPushSubscriptionKey(request.keys())) {
+ sink.error(new InvalidPushSubscriptionKeys(request.keys().get()));
+ }
+ })
+ .then(Mono.defer(() -> pushSubscriptionDAO.save(username, pushSubscription))
+ .thenReturn(pushSubscription));
+ }
+
+ @Override
+ public Mono<PushSubscriptionExpiredTime> updateExpireTime(Username username, PushSubscriptionId id, ZonedDateTime newExpire) {
+ return Mono.just(newExpire)
+ .handle((inputTime, sink) -> {
+ if (newExpire.isBefore(ZonedDateTime.now(clock))) {
+ sink.error(new ExpireTimeInvalidException(inputTime, "expires must be greater than now"));
+ }
+ })
+ .then(getDAO(username).updateExpireTime(username, id, evaluateExpiresTime(Optional.of(newExpire), clock).value())
+ .map(PushSubscriptionExpiredTime::new)
+ .switchIfEmpty(Mono.error(() -> new PushSubscriptionNotFoundException(id))));
+ }
+
+ @Override
+ public Mono<Void> updateTypes(Username username, PushSubscriptionId id, Set<TypeName> types) {
+ return getDAO(username).updateType(username, id, types)
+ .switchIfEmpty(Mono.error(() -> new PushSubscriptionNotFoundException(id)))
+ .then();
+ }
+
+ @Override
+ public Mono<Void> validateVerificationCode(Username username, PushSubscriptionId id) {
+ return getDAO(username)
+ .updateValidated(username, id, true)
+ .switchIfEmpty(Mono.error(() -> new PushSubscriptionNotFoundException(id)))
+ .then();
+ }
+
+ @Override
+ public Mono<Void> revoke(Username username, PushSubscriptionId id) {
+ return getDAO(username).deleteByUsernameAndId(username, id);
+ }
+
+ @Override
+ public Mono<Void> delete(Username username) {
+ return getDAO(username).deleteByUsername(username);
+ }
+
+ @Override
+ public Flux<PushSubscription> get(Username username, Set<PushSubscriptionId> ids) {
+ return getDAO(username).getByUsernameAndIds(username, ids);
+ }
+
+ @Override
+ public Flux<PushSubscription> list(Username username) {
+ return getDAO(username).listByUsername(username);
+ }
+
+ private PostgresPushSubscriptionDAO getDAO(Username username) {
+ return new PostgresPushSubscriptionDAO(executorFactory.create(username.getDomainPart()), typeStateFactory);
+ }
+}
diff --git a/server/data/data-jmap-postgres/src/test/java/org/apache/james/jmap/postgres/pushsubscription/PostgresPushSubscriptionRepositoryTest.java b/server/data/data-jmap-postgres/src/test/java/org/apache/james/jmap/postgres/pushsubscription/PostgresPushSubscriptionRepositoryTest.java
new file mode 100644
index 0000000000..7a471569df
--- /dev/null
+++ b/server/data/data-jmap-postgres/src/test/java/org/apache/james/jmap/postgres/pushsubscription/PostgresPushSubscriptionRepositoryTest.java
@@ -0,0 +1,62 @@
+/****************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one *
+ * or more contributor license agreements. See the NOTICE file *
+ * distributed with this work for additional information *
+ * regarding copyright ownership. The ASF licenses this file *
+ * to you under the Apache License, Version 2.0 (the *
+ * "License"); you may not use this file except in compliance *
+ * with the License. You may obtain a copy of the License at *
+ * *
+ * http://www.apache.org/licenses/LICENSE-2.0 *
+ * *
+ * Unless required by applicable law or agreed to in writing, *
+ * software distributed under the License is distributed on an *
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY *
+ * KIND, either express or implied. See the License for the *
+ * specific language governing permissions and limitations *
+ * under the License. *
+ ****************************************************************/
+
+package org.apache.james.jmap.postgres.pushsubscription;
+
+import java.util.Set;
+
+import org.apache.james.backends.postgres.PostgresExtension;
+import org.apache.james.backends.postgres.PostgresModule;
+import org.apache.james.jmap.api.change.TypeStateFactory;
+import org.apache.james.jmap.api.model.TypeName;
+import org.apache.james.jmap.api.pushsubscription.PushSubscriptionRepository;
+import org.apache.james.jmap.api.pushsubscription.PushSubscriptionRepositoryContract;
+import org.apache.james.utils.UpdatableTickingClock;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import scala.jdk.javaapi.CollectionConverters;
+
+class PostgresPushSubscriptionRepositoryTest implements PushSubscriptionRepositoryContract {
+
+ @RegisterExtension
+ static PostgresExtension postgresExtension = PostgresExtension.withRowLevelSecurity(
+ PostgresModule.aggregateModules(PostgresPushSubscriptionModule.MODULE));
+
+ UpdatableTickingClock clock;
+ PushSubscriptionRepository pushSubscriptionRepository;
+
+ @BeforeEach
+ void setup() {
+ clock = new UpdatableTickingClock(PushSubscriptionRepositoryContract.NOW());
+ pushSubscriptionRepository = new PostgresPushSubscriptionRepository(clock,
+ new TypeStateFactory((Set<TypeName>) CollectionConverters.asJava(PushSubscriptionRepositoryContract.TYPE_NAME_SET())),
+ postgresExtension.getExecutorFactory());
+ }
+
+ @Override
+ public UpdatableTickingClock clock() {
+ return clock;
+ }
+
+ @Override
+ public PushSubscriptionRepository testee() {
+ return pushSubscriptionRepository;
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org