This is an automated email from the ASF dual-hosted git repository. Arsnael pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 4ad14dfb16d16eb964079445c3cbe06d26a916df Author: Benoit TELLIER <[email protected]> AuthorDate: Fri Jul 24 08:14:31 2026 +0200 JAMES-4209 Distributed app content recovery runner --- server/apps/distributed-app/README.adoc | 56 +++++ .../org/apache/james/RecoveryConfiguration.java | 66 ++++++ .../main/java/org/apache/james/S3RecoveryMain.java | 120 ++++++++++ .../java/org/apache/james/S3RecoveryService.java | 252 +++++++++++++++++++++ .../apache/james/RecoveryConfigurationTest.java | 46 ++++ .../org/apache/james/S3RecoveryServiceTest.java | 77 +++++++ 6 files changed, 617 insertions(+) diff --git a/server/apps/distributed-app/README.adoc b/server/apps/distributed-app/README.adoc index 1684f883bd..b81566e744 100644 --- a/server/apps/distributed-app/README.adoc +++ b/server/apps/distributed-app/README.adoc @@ -106,3 +106,59 @@ $ java -Dworking.directory=. -Dlogback.configurationFile=conf/logback.xml -Djdk. Note that binding ports below 1024 requires administrative rights. +== S3 blob store recovery + +When the S3 (or compatible) object store survives but the Cassandra mailbox structure is lost, messages +can be rebuilt from the blob store alone, provided `mailbox.blob.recovery.mode` was set to `synchronous` +or `asynchronous` in `cassandra.properties` while messages were being stored. In that mode James writes, +next to each stored message, a `recovery/<headerBlobId>` sidecar blob holding the matching `bodyBlobId`. + +Recovery is packaged as an alternate main class, `org.apache.james.S3RecoveryMain`, that reuses the same +mailbox, DAO and blob store modules and the same `blobstore.properties` (so AES encryption and +compression are applied transparently). It starts neither the protocol servers nor RabbitMQ. + +It walks the blob store, reads each `recovery/` sidecar, rebuilds the message from its header and body +blobs, reads the `Delivered-To` recipients, and appends the message into a `Restored-messages` mailbox +of each local recipient. + +Run it by overriding the entrypoint main class (Cassandra and S3 must be reachable, RabbitMQ is not +needed): + +[source] +---- +$ java -Dworking.directory=. -Dlogback.configurationFile=conf/logback.xml \ + -cp james-server-distributed-app.jar:james-server-distributed-app.lib/* \ + org.apache.james.S3RecoveryMain +---- + +Or, from the docker image, by overriding the container main class: + +[source] +---- +$ docker run --rm --entrypoint java \ + -v /path/conf:/root/conf \ + apache/james:distributed-latest \ + -Dworking.directory=/root -Dextra.props=/root/conf/jvm.properties \ + -cp '/app/resources:/app/classes:/app/libs/*' \ + org.apache.james.S3RecoveryMain +---- + +An optional date filter restricts recovery to messages whose `Date` header is strictly after the given +ISO-8601 instant. It can be passed as a `--restore-after=<instant>` argument, the `RESTORE_MESSAGES_AFTER` +environment variable, or the `restore.messages.after` system property: + +[source] +---- +$ java ... org.apache.james.S3RecoveryMain --restore-after=2026-01-01T00:00:00Z +---- + +Notes: + +* Restored messages are re-stored (and get a fresh `recovery/` sidecar), so re-running the recovery +restores them again. Restore into an empty deployment, or clean up between runs. +* The search index module is not started during recovery, so restored messages are not indexed on the +fly. Run a re-indexing afterwards if search is needed. + +The same class can be reused for Twake mail by pointing its distributed image at +`org.apache.james.S3RecoveryMain`. + diff --git a/server/apps/distributed-app/src/main/java/org/apache/james/RecoveryConfiguration.java b/server/apps/distributed-app/src/main/java/org/apache/james/RecoveryConfiguration.java new file mode 100644 index 0000000000..1234bf9231 --- /dev/null +++ b/server/apps/distributed-app/src/main/java/org/apache/james/RecoveryConfiguration.java @@ -0,0 +1,66 @@ +/**************************************************************** + * 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; + +import java.time.Instant; +import java.time.format.DateTimeParseException; +import java.util.Arrays; +import java.util.Optional; + +/** + * Configuration for the S3 blob store recovery run. + * + * <p>The optional {@code restoreAfter} instant restricts recovery to messages whose {@code Date} + * header is strictly after the given point in time. It can be provided (highest precedence first) as:</p> + * <ul> + * <li>a {@code --restore-after=<ISO-8601 instant>} program argument</li> + * <li>the {@code RESTORE_MESSAGES_AFTER} environment variable</li> + * <li>the {@code restore.messages.after} system property</li> + * </ul> + */ +public record RecoveryConfiguration(Optional<Instant> restoreAfter) { + private static final String RESTORE_AFTER_ARG = "--restore-after="; + private static final String RESTORE_AFTER_ENV = "RESTORE_MESSAGES_AFTER"; + private static final String RESTORE_AFTER_PROPERTY = "restore.messages.after"; + + public static RecoveryConfiguration parse(String[] args) { + return new RecoveryConfiguration(restoreAfter(args).map(RecoveryConfiguration::parseInstant)); + } + + private static Optional<String> restoreAfter(String[] args) { + return Arrays.stream(args) + .filter(arg -> arg.startsWith(RESTORE_AFTER_ARG)) + .map(arg -> arg.substring(RESTORE_AFTER_ARG.length())) + .findFirst() + .or(() -> Optional.ofNullable(System.getenv(RESTORE_AFTER_ENV))) + .or(() -> Optional.ofNullable(System.getProperty(RESTORE_AFTER_PROPERTY))) + .map(String::trim) + .filter(value -> !value.isEmpty()); + } + + private static Instant parseInstant(String value) { + try { + return Instant.parse(value); + } catch (DateTimeParseException e) { + throw new IllegalArgumentException("Invalid '" + RESTORE_AFTER_ARG + "' value: '" + value + + "'. Expected an ISO-8601 instant, e.g. 2026-01-01T00:00:00Z", e); + } + } +} diff --git a/server/apps/distributed-app/src/main/java/org/apache/james/S3RecoveryMain.java b/server/apps/distributed-app/src/main/java/org/apache/james/S3RecoveryMain.java new file mode 100644 index 0000000000..81019ef18d --- /dev/null +++ b/server/apps/distributed-app/src/main/java/org/apache/james/S3RecoveryMain.java @@ -0,0 +1,120 @@ +/**************************************************************** + * 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; + +import org.apache.james.data.UsersRepositoryModuleChooser; +import org.apache.james.modules.CommonServicesModule; +import org.apache.james.modules.blobstore.BlobStoreCacheModulesChooser; +import org.apache.james.modules.blobstore.BlobStoreConfiguration; +import org.apache.james.modules.blobstore.BlobStoreModulesChooser; +import org.apache.james.modules.data.CassandraSieveQuotaLegacyModule; +import org.apache.james.modules.data.CassandraSieveQuotaModule; +import org.apache.james.modules.data.CassandraUsersRepositoryModule; +import org.apache.james.modules.mailbox.CassandraMailboxQuotaLegacyModule; +import org.apache.james.modules.mailbox.CassandraMailboxQuotaModule; +import org.apache.james.utils.InitializationOperations; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.google.common.collect.ImmutableList; +import com.google.inject.AbstractModule; +import com.google.inject.Guice; +import com.google.inject.Injector; +import com.google.inject.Module; +import com.google.inject.Scopes; +import com.google.inject.util.Modules; + +/** + * Alternate entrypoint for the distributed server that rebuilds messages from the blob store alone. + * + * <p>It reuses the production mailbox, DAO and blob store Guice modules (hence the existing + * {@code blobstore.properties}: AES encryption and compression are applied transparently) but starts + * neither the protocol servers nor RabbitMQ: the in-VM event bus is used because the RabbitMQ override + * of {@link CassandraRabbitMQJamesServerMain} is intentionally not applied.</p> + * + * <p>Run it by overriding the docker entrypoint main class, e.g.:</p> + * <pre> + * docker run --rm --entrypoint java \ + * -v /path/conf:/root/conf \ + * apache/james:distributed-latest \ + * -Dworking.directory=/root -Dextra.props=/root/conf/jvm.properties \ + * -cp '/app/resources:/app/classes:/app/libs/*' \ + * org.apache.james.S3RecoveryMain --restore-after=2026-01-01T00:00:00Z + * </pre> + */ +public class S3RecoveryMain { + private static final Logger LOGGER = LoggerFactory.getLogger(S3RecoveryMain.class); + + public static void main(String[] args) throws Exception { + ExtraProperties.initialize(); + + CassandraRabbitMQJamesConfiguration configuration = CassandraRabbitMQJamesConfiguration.builder() + .useWorkingDirectoryEnvProperty() + .build(); + RecoveryConfiguration recoveryConfiguration = RecoveryConfiguration.parse(args); + + LOGGER.info("Loading configuration {}", configuration); + Injector injector = Guice.createInjector(recoveryModule(configuration, recoveryConfiguration)); + injector.getInstance(InitializationOperations.class).initModules(); + + S3RecoveryService.Report report = injector.getInstance(S3RecoveryService.class).run().block(); + LOGGER.info("S3 recovery completed: {}", report); + + System.exit(0); + } + + private static Module recoveryModule(CassandraRabbitMQJamesConfiguration configuration, + RecoveryConfiguration recoveryConfiguration) { + BlobStoreConfiguration blobStoreConfiguration = configuration.blobStoreConfiguration(); + + return Modules.combine(ImmutableList.<Module>builder() + .add(new CommonServicesModule(configuration)) + .add(CassandraRabbitMQJamesServerMain.CASSANDRA_SERVER_CORE_MODULE) + .add(CassandraRabbitMQJamesServerMain.CASSANDRA_MAILBOX_MODULE) + .add(chooseQuotaModule(configuration)) + .addAll(new UsersRepositoryModuleChooser(new CassandraUsersRepositoryModule()) + .chooseModules(configuration.getUsersRepositoryImplementation())) + .addAll(BlobStoreModulesChooser.chooseModules(blobStoreConfiguration)) + .addAll(BlobStoreCacheModulesChooser.chooseModules(blobStoreConfiguration)) + .add(new RecoveryModule(recoveryConfiguration)) + .build()); + } + + private static Module chooseQuotaModule(CassandraRabbitMQJamesConfiguration configuration) { + if (configuration.isQuotaCompatibilityMode()) { + return Modules.combine(new CassandraMailboxQuotaLegacyModule(), new CassandraSieveQuotaLegacyModule()); + } + return Modules.combine(new CassandraMailboxQuotaModule(), new CassandraSieveQuotaModule()); + } + + static class RecoveryModule extends AbstractModule { + private final RecoveryConfiguration recoveryConfiguration; + + RecoveryModule(RecoveryConfiguration recoveryConfiguration) { + this.recoveryConfiguration = recoveryConfiguration; + } + + @Override + protected void configure() { + bind(RecoveryConfiguration.class).toInstance(recoveryConfiguration); + bind(S3RecoveryService.class).in(Scopes.SINGLETON); + } + } +} diff --git a/server/apps/distributed-app/src/main/java/org/apache/james/S3RecoveryService.java b/server/apps/distributed-app/src/main/java/org/apache/james/S3RecoveryService.java new file mode 100644 index 0000000000..703ddc10ce --- /dev/null +++ b/server/apps/distributed-app/src/main/java/org/apache/james/S3RecoveryService.java @@ -0,0 +1,252 @@ +/**************************************************************** + * 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; + +import static org.apache.james.blob.api.BlobStore.StoragePolicy.LOW_COST; +import static org.apache.james.blob.api.BlobStore.StoragePolicy.SIZE_BASED; +import static org.apache.james.blob.api.BlobStoreDAO.RECOVERY_BLOB_PREFIX; + +import java.io.ByteArrayInputStream; +import java.nio.charset.StandardCharsets; +import java.util.Date; +import java.util.List; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; + +import jakarta.inject.Inject; + +import org.apache.james.blob.api.BlobId; +import org.apache.james.blob.api.BlobStore; +import org.apache.james.blob.api.BlobStoreDAO; +import org.apache.james.blob.api.BucketName; +import org.apache.james.core.MailAddress; +import org.apache.james.core.Username; +import org.apache.james.mailbox.MailboxManager; +import org.apache.james.mailbox.MailboxSession; +import org.apache.james.mailbox.MessageManager; +import org.apache.james.mailbox.exception.MailboxExistsException; +import org.apache.james.mailbox.model.Content; +import org.apache.james.mailbox.model.HeaderAndBodyByteContent; +import org.apache.james.mailbox.model.MailboxPath; +import org.apache.james.mime4j.dom.Message; +import org.apache.james.mime4j.message.DefaultMessageBuilder; +import org.apache.james.mime4j.stream.Field; +import org.apache.james.mime4j.stream.MimeConfig; +import org.apache.james.user.api.UsersRepository; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.google.common.annotations.VisibleForTesting; + +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.core.scheduler.Schedulers; + +/** + * Walks the blob store looking for {@code recovery/} sidecars (written by + * {@code CassandraMessageDAOV3} when {@code mailbox.blob.recovery.mode} is enabled) and restores the + * associated messages into a {@code Restored-messages} mailbox of each local {@code Delivered-To} recipient. + * + * <p>Reads go through the configured {@link BlobStore}, so AES decryption and decompression are applied + * transparently, exactly as the write path did.</p> + */ +public class S3RecoveryService { + public record Report(long processed, long restored, long skippedByDate, long skippedNoLocalUser, long failed) { + public static Report empty() { + return new Report(0, 0, 0, 0, 0); + } + + Report merge(Report other) { + return new Report(processed + other.processed, + restored + other.restored, + skippedByDate + other.skippedByDate, + skippedNoLocalUser + other.skippedNoLocalUser, + failed + other.failed); + } + } + + private record RecoveredMessage(List<MailAddress> recipients, Optional<Date> date, Content content) { + } + + private static final Logger LOGGER = LoggerFactory.getLogger(S3RecoveryService.class); + private static final String RESTORE_MAILBOX = "Restored-messages"; + private static final int CONCURRENCY = 8; + private static final String DELIVERED_TO = "Delivered-To"; + private static final Report RESTORED = new Report(1, 1, 0, 0, 0); + private static final Report SKIPPED_BY_DATE = new Report(1, 0, 1, 0, 0); + private static final Report SKIPPED_NO_LOCAL_USER = new Report(1, 0, 0, 1, 0); + private static final Report FAILED = new Report(1, 0, 0, 0, 1); + + private final BlobStore blobStore; + private final BlobStoreDAO blobStoreDAO; + private final BlobId.Factory blobIdFactory; + private final MailboxManager mailboxManager; + private final UsersRepository usersRepository; + private final RecoveryConfiguration configuration; + private final Set<Username> ensuredMailboxes = ConcurrentHashMap.newKeySet(); + + @Inject + public S3RecoveryService(BlobStore blobStore, BlobStoreDAO blobStoreDAO, BlobId.Factory blobIdFactory, + MailboxManager mailboxManager, UsersRepository usersRepository, + RecoveryConfiguration configuration) { + this.blobStore = blobStore; + this.blobStoreDAO = blobStoreDAO; + this.blobIdFactory = blobIdFactory; + this.mailboxManager = mailboxManager; + this.usersRepository = usersRepository; + this.configuration = configuration; + } + + public Mono<Report> run() { + BucketName bucket = blobStore.getDefaultBucketName(); + LOGGER.info("Starting S3 recovery on bucket {} (restore after: {})", bucket.asString(), configuration.restoreAfter()); + return Flux.from(blobStoreDAO.listBlobs(bucket)) + .map(BlobId::asString) + .filter(key -> key.startsWith(RECOVERY_BLOB_PREFIX)) + .flatMap(recoveryKey -> restoreOne(bucket, recoveryKey), CONCURRENCY) + .reduce(Report.empty(), Report::merge) + .doOnNext(report -> LOGGER.info("S3 recovery finished: {}", report)); + } + + private Mono<Report> restoreOne(BucketName bucket, String recoveryKey) { + String headerKey = recoveryKey.substring(RECOVERY_BLOB_PREFIX.length()); + BlobId headerBlobId = blobIdFactory.parse(headerKey); + BlobId recoveryBlobId = blobIdFactory.parse(recoveryKey); + + return Mono.from(blobStoreDAO.readBytes(bucket, recoveryBlobId)) + .map(sidecar -> blobIdFactory.parse(new String(sidecar.payload(), StandardCharsets.UTF_8).trim())) + .flatMap(bodyBlobId -> recover(bucket, headerBlobId, bodyBlobId)) + .flatMap(this::restore) + .onErrorResume(error -> { + LOGGER.error("Failed to recover message from {}", recoveryKey, error); + return Mono.just(FAILED); + }); + } + + private Mono<RecoveredMessage> recover(BucketName bucket, BlobId headerBlobId, BlobId bodyBlobId) { + return Mono.from(blobStore.readBytes(bucket, headerBlobId, SIZE_BASED)) + .flatMap(headerBytes -> { + MessageHeaders headers = parseHeaders(headerBytes); + return Mono.from(blobStore.readBytes(bucket, bodyBlobId, LOW_COST)) + .map(bodyBytes -> new RecoveredMessage(headers.recipients(), headers.date(), + new HeaderAndBodyByteContent(headerBytes, bodyBytes))); + }); + } + + private Mono<Report> restore(RecoveredMessage message) { + if (filteredOutByDate(message.date())) { + return Mono.just(SKIPPED_BY_DATE); + } + return Flux.fromIterable(message.recipients()) + .concatMap(recipient -> localUser(recipient).flux()) + .collectList() + .flatMap(users -> { + if (users.isEmpty()) { + LOGGER.info("Skipping message: no local Delivered-To recipient among {}", message.recipients()); + return Mono.just(SKIPPED_NO_LOCAL_USER); + } + return Flux.fromIterable(users) + .concatMap(user -> appendToRestored(user, message)) + .then(Mono.just(RESTORED)); + }); + } + + private boolean filteredOutByDate(Optional<Date> date) { + return configuration.restoreAfter() + .flatMap(after -> date.map(value -> !value.toInstant().isAfter(after))) + .orElse(false); + } + + private Mono<Username> localUser(MailAddress recipient) { + return Mono.fromCallable(() -> usersRepository.getUsername(recipient)) + .flatMap(username -> Mono.from(usersRepository.containsReactive(username)) + .<Username>handle((present, sink) -> { + if (present) { + sink.next(username); + } + })) + .onErrorResume(error -> { + LOGGER.warn("Unable to resolve local user for {}", recipient.asString(), error); + return Mono.empty(); + }); + } + + private Mono<Void> appendToRestored(Username user, RecoveredMessage message) { + return Mono.fromRunnable(() -> doAppend(user, message)) + .subscribeOn(Schedulers.boundedElastic()) + .then(); + } + + private void doAppend(Username user, RecoveredMessage message) { + try { + MailboxSession session = mailboxManager.createSystemSession(user); + MailboxPath path = ensureMailbox(user, session); + MessageManager messageManager = mailboxManager.getMailbox(path, session); + messageManager.appendMessage(MessageManager.AppendCommand.builder() + .withInternalDate(message.date()) + .notRecent() + .build(message.content()), session); + } catch (Exception e) { + throw new RuntimeException("Failed to append recovered message to " + user.asString(), e); + } + } + + private MailboxPath ensureMailbox(Username user, MailboxSession session) throws Exception { + MailboxPath path = MailboxPath.forUser(user, RESTORE_MAILBOX); + if (ensuredMailboxes.add(user)) { + try { + mailboxManager.createMailbox(path, session); + } catch (MailboxExistsException e) { + // The mailbox already exists, nothing to do. + } + } + return path; + } + + @VisibleForTesting + static MessageHeaders parseHeaders(byte[] headerBytes) { + try { + DefaultMessageBuilder messageBuilder = new DefaultMessageBuilder(); + messageBuilder.setMimeEntityConfig(MimeConfig.PERMISSIVE); + Message message = messageBuilder.parseMessage(new ByteArrayInputStream(headerBytes)); + List<MailAddress> recipients = message.getHeader().getFields(DELIVERED_TO).stream() + .map(Field::getBody) + .flatMap(value -> toMailAddress(value).stream()) + .toList(); + return new MessageHeaders(recipients, Optional.ofNullable(message.getDate())); + } catch (Exception e) { + LOGGER.warn("Unable to parse recovered message headers", e); + return new MessageHeaders(List.of(), Optional.empty()); + } + } + + private static Optional<MailAddress> toMailAddress(String value) { + try { + return Optional.of(new MailAddress(value.trim())); + } catch (Exception e) { + LOGGER.warn("Unable to parse Delivered-To value '{}'", value); + return Optional.empty(); + } + } + + record MessageHeaders(List<MailAddress> recipients, Optional<Date> date) { + } +} diff --git a/server/apps/distributed-app/src/test/java/org/apache/james/RecoveryConfigurationTest.java b/server/apps/distributed-app/src/test/java/org/apache/james/RecoveryConfigurationTest.java new file mode 100644 index 0000000000..8fdd7f0a37 --- /dev/null +++ b/server/apps/distributed-app/src/test/java/org/apache/james/RecoveryConfigurationTest.java @@ -0,0 +1,46 @@ +/**************************************************************** + * 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; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.time.Instant; + +import org.junit.jupiter.api.Test; + +class RecoveryConfigurationTest { + @Test + void parseShouldReturnEmptyWhenNoDateProvided() { + assertThat(RecoveryConfiguration.parse(new String[] {}).restoreAfter()).isEmpty(); + } + + @Test + void parseShouldReadRestoreAfterArgument() { + assertThat(RecoveryConfiguration.parse(new String[] {"--restore-after=2026-01-01T00:00:00Z"}).restoreAfter()) + .contains(Instant.parse("2026-01-01T00:00:00Z")); + } + + @Test + void parseShouldRejectInvalidInstant() { + assertThatThrownBy(() -> RecoveryConfiguration.parse(new String[] {"--restore-after=not-a-date"})) + .isInstanceOf(IllegalArgumentException.class); + } +} diff --git a/server/apps/distributed-app/src/test/java/org/apache/james/S3RecoveryServiceTest.java b/server/apps/distributed-app/src/test/java/org/apache/james/S3RecoveryServiceTest.java new file mode 100644 index 0000000000..067f1e9349 --- /dev/null +++ b/server/apps/distributed-app/src/test/java/org/apache/james/S3RecoveryServiceTest.java @@ -0,0 +1,77 @@ +/**************************************************************** + * 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; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.nio.charset.StandardCharsets; +import java.time.Instant; + +import org.apache.james.core.MailAddress; +import org.junit.jupiter.api.Test; + +class S3RecoveryServiceTest { + @Test + void parseHeadersShouldExtractEveryDeliveredToRecipient() { + byte[] headers = ("Delivered-To: [email protected]\r\n" + + "Delivered-To: [email protected]\r\n" + + "Subject: hello\r\n" + + "\r\n").getBytes(StandardCharsets.UTF_8); + + assertThat(S3RecoveryService.parseHeaders(headers).recipients()) + .extracting(MailAddress::asString) + .containsExactly("[email protected]", "[email protected]"); + } + + @Test + void parseHeadersShouldExtractDate() { + byte[] headers = ("Delivered-To: [email protected]\r\n" + + "Date: Wed, 01 Jan 2020 00:00:00 +0000\r\n" + + "\r\n").getBytes(StandardCharsets.UTF_8); + + assertThat(S3RecoveryService.parseHeaders(headers).date()) + .hasValueSatisfying(date -> assertThat(date.toInstant()).isEqualTo(Instant.parse("2020-01-01T00:00:00Z"))); + } + + @Test + void parseHeadersShouldReturnNoRecipientWhenNoDeliveredTo() { + byte[] headers = ("Subject: hello\r\n\r\n").getBytes(StandardCharsets.UTF_8); + + assertThat(S3RecoveryService.parseHeaders(headers).recipients()).isEmpty(); + } + + @Test + void parseHeadersShouldReturnNoDateWhenAbsent() { + byte[] headers = ("Delivered-To: [email protected]\r\n\r\n").getBytes(StandardCharsets.UTF_8); + + assertThat(S3RecoveryService.parseHeaders(headers).date()).isEmpty(); + } + + @Test + void parseHeadersShouldSkipUnparseableDeliveredToValue() { + byte[] headers = ("Delivered-To: not-an-address\r\n" + + "Delivered-To: [email protected]\r\n" + + "\r\n").getBytes(StandardCharsets.UTF_8); + + assertThat(S3RecoveryService.parseHeaders(headers).recipients()) + .extracting(MailAddress::asString) + .containsExactly("[email protected]"); + } +} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
