Skip to content

James 4150 #2919

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
chibenwa merged 4 commits into from
Jan 29, 2026
Merged

James 4150 #2919

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
0e767e2
JAMES-4150 Make Attachment Storage Optional for PostgreSQL Message St…
omerfarukicen Jan 26, 2026
d407de6
JAMES-4150 Make Attachment Storage Optional for PostgreSQL Message St…
omerfarukicen Jan 26, 2026
243424e
fix rebase problem
omerfarukicen Jan 26, 2026
c016742
fix checkstyle
omerfarukicen Jan 26, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
102 changes: 62 additions & 40 deletions ...mmon/postgres/src/main/java/org/apache/james/backends/postgres/PostgresConfiguration.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,8 @@ public class PostgresConfiguration {
public static final String SSL_MODE_DEFAULT_VALUE = "allow";
public static final String JOOQ_REACTIVE_TIMEOUT = "jooq.reactive.timeout";
public static final Duration JOOQ_REACTIVE_TIMEOUT_DEFAULT_VALUE = Duration.ofSeconds(10);
public static final String ATTACHMENT_STORAGE_ENABLED = "attachment.storage.enabled";
public static final boolean ATTACHMENT_STORAGE_ENABLED_DEFAULT_VALUE = true;

public static class Credential {
private final String username;
Expand Down Expand Up @@ -95,6 +97,7 @@ public static class Builder {
private Optional<Integer> byPassRLSPoolMaxSize = Optional.empty();
private Optional<String> sslMode = Optional.empty();
private Optional<Duration> jooqReactiveTimeout = Optional.empty();
private Optional<Boolean> attachmentStorageEnabled = Optional.empty();

public Builder databaseName(String databaseName) {
this.databaseName = Optional.of(databaseName);
Expand Down Expand Up @@ -241,6 +244,16 @@ public Builder jooqReactiveTimeout(Optional<Duration> jooqReactiveTimeout) {
return this;
}

public Builder attachmentStorageEnabled(Optional<Boolean> attachmentStorageEnabled) {
this.attachmentStorageEnabled = attachmentStorageEnabled;
return this;
}

public Builder attachmentStorageEnabled(Boolean attachmentStorageEnabled) {
this.attachmentStorageEnabled = Optional.of(attachmentStorageEnabled);
return this;
}

public PostgresConfiguration build() {
Preconditions.checkArgument(username.isPresent() && !username.get().isBlank(), "You need to specify username");
Preconditions.checkArgument(password.isPresent() && !password.get().isBlank(), "You need to specify password");
Expand All @@ -251,18 +264,19 @@ public PostgresConfiguration build() {
}

return new PostgresConfiguration(host.orElse(HOST_DEFAULT_VALUE),
port.orElse(PORT_DEFAULT_VALUE),
databaseName.orElse(DATABASE_NAME_DEFAULT_VALUE),
databaseSchema.orElse(DATABASE_SCHEMA_DEFAULT_VALUE),
new Credential(username.get(), password.get()),
new Credential(byPassRLSUser.orElse(username.get()), byPassRLSPassword.orElse(password.get())),
port.orElse(PORT_DEFAULT_VALUE),
databaseName.orElse(DATABASE_NAME_DEFAULT_VALUE),
databaseSchema.orElse(DATABASE_SCHEMA_DEFAULT_VALUE),
new Credential(username.get(), password.get()),
new Credential(byPassRLSUser.orElse(username.get()), byPassRLSPassword.orElse(password.get())),
rowLevelSecurityEnabled.filter(rlsEnabled -> rlsEnabled).map(rlsEnabled -> RowLevelSecurity.ENABLED).orElse(RowLevelSecurity.DISABLED),
poolInitialSize.orElse(POOL_INITIAL_SIZE_DEFAULT_VALUE),
poolMaxSize.orElse(POOL_MAX_SIZE_DEFAULT_VALUE),
byPassRLSPoolInitialSize.orElse(BY_PASS_RLS_POOL_INITIAL_SIZE_DEFAULT_VALUE),
byPassRLSPoolMaxSize.orElse(BY_PASS_RLS_POOL_MAX_SIZE_DEFAULT_VALUE),
SSLMode.fromValue(sslMode.orElse(SSL_MODE_DEFAULT_VALUE)),
jooqReactiveTimeout.orElse(JOOQ_REACTIVE_TIMEOUT_DEFAULT_VALUE));
poolInitialSize.orElse(POOL_INITIAL_SIZE_DEFAULT_VALUE),
poolMaxSize.orElse(POOL_MAX_SIZE_DEFAULT_VALUE),
byPassRLSPoolInitialSize.orElse(BY_PASS_RLS_POOL_INITIAL_SIZE_DEFAULT_VALUE),
byPassRLSPoolMaxSize.orElse(BY_PASS_RLS_POOL_MAX_SIZE_DEFAULT_VALUE),
SSLMode.fromValue(sslMode.orElse(SSL_MODE_DEFAULT_VALUE)),
jooqReactiveTimeout.orElse(JOOQ_REACTIVE_TIMEOUT_DEFAULT_VALUE),
attachmentStorageEnabled.orElse(ATTACHMENT_STORAGE_ENABLED_DEFAULT_VALUE));
}
}

Expand All @@ -272,23 +286,24 @@ public static Builder builder() {

public static PostgresConfiguration from(Configuration propertiesConfiguration) {
return builder()
.databaseName(Optional.ofNullable(propertiesConfiguration.getString(DATABASE_NAME)))
.databaseSchema(Optional.ofNullable(propertiesConfiguration.getString(DATABASE_SCHEMA)))
.host(Optional.ofNullable(propertiesConfiguration.getString(HOST)))
.port(propertiesConfiguration.getInt(PORT, PORT_DEFAULT_VALUE))
.username(Optional.ofNullable(propertiesConfiguration.getString(USERNAME)))
.password(Optional.ofNullable(propertiesConfiguration.getString(PASSWORD)))
.byPassRLSUser(Optional.ofNullable(propertiesConfiguration.getString(BY_PASS_RLS_USERNAME)))
.byPassRLSPassword(Optional.ofNullable(propertiesConfiguration.getString(BY_PASS_RLS_PASSWORD)))
.rowLevelSecurityEnabled(propertiesConfiguration.getBoolean(RLS_ENABLED, false))
.poolInitialSize(Optional.ofNullable(propertiesConfiguration.getInteger(POOL_INITIAL_SIZE, null)))
.poolMaxSize(Optional.ofNullable(propertiesConfiguration.getInteger(POOL_MAX_SIZE, null)))
.byPassRLSPoolInitialSize(Optional.ofNullable(propertiesConfiguration.getInteger(BY_PASS_RLS_POOL_INITIAL_SIZE, null)))
.byPassRLSPoolMaxSize(Optional.ofNullable(propertiesConfiguration.getInteger(BY_PASS_RLS_POOL_MAX_SIZE, null)))
.sslMode(Optional.ofNullable(propertiesConfiguration.getString(SSL_MODE)))
.jooqReactiveTimeout(Optional.ofNullable(propertiesConfiguration.getString(JOOQ_REACTIVE_TIMEOUT))
.map(value -> DurationParser.parse(value, ChronoUnit.SECONDS)))
.build();
.databaseName(Optional.ofNullable(propertiesConfiguration.getString(DATABASE_NAME)))
.databaseSchema(Optional.ofNullable(propertiesConfiguration.getString(DATABASE_SCHEMA)))
.host(Optional.ofNullable(propertiesConfiguration.getString(HOST)))
.port(propertiesConfiguration.getInt(PORT, PORT_DEFAULT_VALUE))
.username(Optional.ofNullable(propertiesConfiguration.getString(USERNAME)))
.password(Optional.ofNullable(propertiesConfiguration.getString(PASSWORD)))
.byPassRLSUser(Optional.ofNullable(propertiesConfiguration.getString(BY_PASS_RLS_USERNAME)))
.byPassRLSPassword(Optional.ofNullable(propertiesConfiguration.getString(BY_PASS_RLS_PASSWORD)))
.rowLevelSecurityEnabled(propertiesConfiguration.getBoolean(RLS_ENABLED, false))
.poolInitialSize(Optional.ofNullable(propertiesConfiguration.getInteger(POOL_INITIAL_SIZE, null)))
.poolMaxSize(Optional.ofNullable(propertiesConfiguration.getInteger(POOL_MAX_SIZE, null)))
.byPassRLSPoolInitialSize(Optional.ofNullable(propertiesConfiguration.getInteger(BY_PASS_RLS_POOL_INITIAL_SIZE, null)))
.byPassRLSPoolMaxSize(Optional.ofNullable(propertiesConfiguration.getInteger(BY_PASS_RLS_POOL_MAX_SIZE, null)))
.sslMode(Optional.ofNullable(propertiesConfiguration.getString(SSL_MODE)))
.jooqReactiveTimeout(Optional.ofNullable(propertiesConfiguration.getString(JOOQ_REACTIVE_TIMEOUT))
.map(value -> DurationParser.parse(value, ChronoUnit.SECONDS)))
.attachmentStorageEnabled(propertiesConfiguration.getBoolean(ATTACHMENT_STORAGE_ENABLED, ATTACHMENT_STORAGE_ENABLED_DEFAULT_VALUE))
.build();
}

private final String host;
Expand All @@ -304,12 +319,13 @@ public static PostgresConfiguration from(Configuration propertiesConfiguration)
private final Integer byPassRLSPoolMaxSize;
private final SSLMode sslMode;
private final Duration jooqReactiveTimeout;
private final boolean attachmentStorageEnabled;

private PostgresConfiguration(String host, int port, String databaseName, String databaseSchema,
Credential defaultCredential, Credential byPassRLSCredential, RowLevelSecurity rowLevelSecurity,
Integer poolInitialSize, Integer poolMaxSize,
Integer byPassRLSPoolInitialSize, Integer byPassRLSPoolMaxSize,
SSLMode sslMode, Duration jooqReactiveTimeout) {
SSLMode sslMode, Duration jooqReactiveTimeout, boolean attachmentStorageEnabled) {
this.host = host;
this.port = port;
this.databaseName = databaseName;
Expand All @@ -323,6 +339,7 @@ private PostgresConfiguration(String host, int port, String databaseName, String
this.byPassRLSPoolMaxSize = byPassRLSPoolMaxSize;
this.sslMode = sslMode;
this.jooqReactiveTimeout = jooqReactiveTimeout;
this.attachmentStorageEnabled = attachmentStorageEnabled;
}

public String getHost() {
Expand Down Expand Up @@ -377,9 +394,13 @@ public Duration getJooqReactiveTimeout() {
return jooqReactiveTimeout;
}

public boolean isAttachmentStorageEnabled() {
return attachmentStorageEnabled;
}

@Override
public final int hashCode() {
return Objects.hash(host, port, databaseName, databaseSchema, defaultCredential, byPassRLSCredential, rowLevelSecurity, poolInitialSize, poolMaxSize, sslMode, jooqReactiveTimeout);
return Objects.hash(host, port, databaseName, databaseSchema, defaultCredential, byPassRLSCredential, rowLevelSecurity, poolInitialSize, poolMaxSize, sslMode, jooqReactiveTimeout, attachmentStorageEnabled);
}

@Override
Expand All @@ -388,16 +409,17 @@ public final boolean equals(Object o) {
PostgresConfiguration that = (PostgresConfiguration) o;

return Objects.equals(this.rowLevelSecurity, that.rowLevelSecurity)
&& Objects.equals(this.host, that.host)
&& Objects.equals(this.port, that.port)
&& Objects.equals(this.defaultCredential, that.defaultCredential)
&& Objects.equals(this.byPassRLSCredential, that.byPassRLSCredential)
&& Objects.equals(this.databaseName, that.databaseName)
&& Objects.equals(this.databaseSchema, that.databaseSchema)
&& Objects.equals(this.poolInitialSize, that.poolInitialSize)
&& Objects.equals(this.poolMaxSize, that.poolMaxSize)
&& Objects.equals(this.sslMode, that.sslMode)
&& Objects.equals(this.jooqReactiveTimeout, that.jooqReactiveTimeout);
&& Objects.equals(this.host, that.host)
&& Objects.equals(this.port, that.port)
&& Objects.equals(this.defaultCredential, that.defaultCredential)
&& Objects.equals(this.byPassRLSCredential, that.byPassRLSCredential)
&& Objects.equals(this.databaseName, that.databaseName)
&& Objects.equals(this.databaseSchema, that.databaseSchema)
&& Objects.equals(this.poolInitialSize, that.poolInitialSize)
&& Objects.equals(this.poolMaxSize, that.poolMaxSize)
&& Objects.equals(this.sslMode, that.sslMode)
&& Objects.equals(this.jooqReactiveTimeout, that.jooqReactiveTimeout)
&& Objects.equals(this.attachmentStorageEnabled, that.attachmentStorageEnabled);
}
return false;
}
Expand Down
15 changes: 13 additions & 2 deletions mailbox/postgres/src/main/java/org/apache/james/mailbox/postgres/DeleteMessageListener.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import jakarta.inject.Named;

import org.apache.commons.io.IOUtils;
import org.apache.james.backends.postgres.PostgresConfiguration;
import org.apache.james.blob.api.BlobStore;
import org.apache.james.core.Username;
import org.apache.james.events.Event;
Expand Down Expand Up @@ -64,20 +65,23 @@ public static class DeleteMessageListenerGroup extends Group {
private final PostgresAttachmentDAO.Factory attachmentDAOFactory;
private final PostgresThreadDAO.Factory threadDAOFactory;
private final EventBus contentDeletionEventBus;
private final PostgresConfiguration postgresConfiguration;

@Inject
public DeleteMessageListener(BlobStore blobStore,
PostgresMailboxMessageDAO.Factory mailboxMessageDAOFactory,
PostgresMessageDAO.Factory messageDAOFactory,
PostgresAttachmentDAO.Factory attachmentDAOFactory,
PostgresThreadDAO.Factory threadDAOFactory,
@Named(CONTENT_DELETION) EventBus contentDeletionEventBus) {
@Named(CONTENT_DELETION) EventBus contentDeletionEventBus,
PostgresConfiguration postgresConfiguration) {
this.messageDAOFactory = messageDAOFactory;
this.mailboxMessageDAOFactory = mailboxMessageDAOFactory;
this.blobStore = blobStore;
this.attachmentDAOFactory = attachmentDAOFactory;
this.threadDAOFactory = threadDAOFactory;
this.contentDeletionEventBus = contentDeletionEventBus;
this.postgresConfiguration = postgresConfiguration;
}

@Override
Expand Down Expand Up @@ -141,7 +145,7 @@ private Mono<Void> handleMessageDeletion(PostgresMessageDAO postgresMessageDAO,
.flatMap(msgId -> postgresMessageDAO.retrieveMessage(messageId)
.flatMap(messageRepresentation -> dispatchMessageContentDeletionEvent(mailboxId, owner, messageRepresentation))
.then(deleteBodyBlob(msgId, postgresMessageDAO))
.then(deleteAttachment(msgId, attachmentDAO))
.then(deleteAttachmentIfEnabled(msgId, attachmentDAO))
.then(threadDAO.deleteSome(owner, msgId))
.then(postgresMessageDAO.deleteByMessageId(msgId)));
}
Expand Down Expand Up @@ -188,4 +192,11 @@ private Mono<Void> deleteAttachmentBlobs(PostgresMessageId messageId, PostgresAt
.flatMap(blobId -> Mono.from(blobStore.delete(blobStore.getDefaultBucketName(), blobId)), ReactorUtils.DEFAULT_CONCURRENCY)
.then();
}

private Mono<Void> deleteAttachmentIfEnabled(PostgresMessageId messageId, PostgresAttachmentDAO attachmentDAO) {
if (postgresConfiguration.isAttachmentStorageEnabled()) {
return deleteAttachment(messageId, attachmentDAO);
}
return Mono.empty();
}
}
11 changes: 9 additions & 2 deletions .../src/main/java/org/apache/james/mailbox/postgres/PostgresMailboxSessionMapperFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,8 @@ public class PostgresMailboxSessionMapperFactory extends MailboxSessionMapperFac
private final Clock clock;
private final RowLevelSecurity rowLevelSecurity;
private final AttachmentIdAssignationStrategy attachmentIdAssignationStrategy;
private final boolean attachmentStorageEnabled;
private final PostgresConfiguration postgresConfiguration;

@Inject
public PostgresMailboxSessionMapperFactory(PostgresExecutor.Factory executorFactory,
Expand All @@ -78,7 +80,9 @@ public PostgresMailboxSessionMapperFactory(PostgresExecutor.Factory executorFact
this.blobIdFactory = blobIdFactory;
this.clock = clock;
this.rowLevelSecurity = postgresConfiguration.getRowLevelSecurity();
this.attachmentStorageEnabled = postgresConfiguration.isAttachmentStorageEnabled();
this.attachmentIdAssignationStrategy = attachmentIdAssignationStrategy;
this.postgresConfiguration = postgresConfiguration;
}

@Override
Expand Down Expand Up @@ -146,14 +150,17 @@ public PostgresAttachmentMapper getAttachmentMapper(MailboxSession session) {
return createAttachmentMapper(session);
}

public boolean isAttachmentStorageEnabled() {
return attachmentStorageEnabled;
}

@VisibleForTesting
protected DeleteMessageListener deleteMessageListener(EventBus contentDeletionEventBus) {
PostgresMessageDAO.Factory postgresMessageDAOFactory = new PostgresMessageDAO.Factory(blobIdFactory, executorFactory);
PostgresMailboxMessageDAO.Factory postgresMailboxMessageDAOFactory = new PostgresMailboxMessageDAO.Factory(executorFactory);
PostgresAttachmentDAO.Factory attachmentDAOFactory = new PostgresAttachmentDAO.Factory(executorFactory, blobIdFactory);
PostgresThreadDAO.Factory threadDAOFactory = new PostgresThreadDAO.Factory(executorFactory);

return new DeleteMessageListener(blobStore, postgresMailboxMessageDAOFactory, postgresMessageDAOFactory,
attachmentDAOFactory, threadDAOFactory, contentDeletionEventBus);
attachmentDAOFactory, threadDAOFactory, contentDeletionEventBus, postgresConfiguration);
}
}
15 changes: 13 additions & 2 deletions mailbox/postgres/src/main/java/org/apache/james/mailbox/postgres/PostgresMessageManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,13 +69,24 @@ public PostgresMessageManager(PostgresMailboxSessionMapperFactory mapperFactory,
StoreRightManager storeRightManager, ThreadIdGuessingAlgorithm threadIdGuessingAlgorithm,
Clock clock, PreDeletionHooks preDeletionHooks) {
super(StoreMailboxManager.DEFAULT_NO_MESSAGE_CAPABILITIES, mapperFactory, index, eventBus, locker, mailbox,
quotaManager, quotaRootResolver, batchSizes, storeRightManager, preDeletionHooks,
new MessageStorer.WithAttachment(mapperFactory, messageIdFactory, new MessageFactory.StoreMessageFactory(), mapperFactory, messageParser, threadIdGuessingAlgorithm, clock));
quotaManager, quotaRootResolver, batchSizes, storeRightManager, preDeletionHooks,
createMessageStorer(mapperFactory, messageIdFactory, messageParser, threadIdGuessingAlgorithm, clock));
this.storeRightManager = storeRightManager;
this.mapperFactory = mapperFactory;
this.mailbox = mailbox;
}

private static MessageStorer createMessageStorer(PostgresMailboxSessionMapperFactory mapperFactory,
MessageId.Factory messageIdFactory,
MessageParser messageParser,
ThreadIdGuessingAlgorithm threadIdGuessingAlgorithm,
Clock clock) {
if (mapperFactory.isAttachmentStorageEnabled()) {
return new MessageStorer.WithAttachment(mapperFactory, messageIdFactory, new MessageFactory.StoreMessageFactory(), mapperFactory, messageParser, threadIdGuessingAlgorithm, clock);
} else {
return new MessageStorer.WithoutAttachment(mapperFactory, messageIdFactory, new MessageFactory.StoreMessageFactory(), threadIdGuessingAlgorithm, clock);
}
}

Copy link
Contributor

@chibenwa chibenwa Jan 26, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We are missing the critical changes of mailbox/postgres/src/main/java/org/apache/james/mailbox/postgres/PostgresMessageManager.java where we actually chose which messageStorer we want: I suspect a rebase issue.

Copy link
Contributor Author

@omerfarukicen omerfarukicen Jan 26, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes Rebase problem .

@Override
public Flags getPermanentFlags(MailboxSession session) {
Expand Down
4 changes: 3 additions & 1 deletion ...src/test/java/org/apache/james/mailbox/postgres/PostgresMailboxManagerAttachmentTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
import java.time.Clock;
import java.time.Instant;

import org.apache.james.backends.postgres.PostgresConfiguration;
import org.apache.james.backends.postgres.PostgresExtension;
import org.apache.james.blob.api.BlobId;
import org.apache.james.blob.api.BucketName;
Expand Down Expand Up @@ -82,6 +83,7 @@ public class PostgresMailboxManagerAttachmentTest extends AbstractMailboxManager
void beforeAll() throws Exception {
BlobId.Factory blobIdFactory = new PlainBlobId.Factory();
DeDuplicationBlobStore blobStore = new DeDuplicationBlobStore(new MemoryBlobStoreDAO(), BucketName.DEFAULT, blobIdFactory);
PostgresConfiguration postgresConfiguration = PostgresConfiguration.builder().username("a").password("a").build();
mapperFactory = new PostgresMailboxSessionMapperFactory(postgresExtension.getExecutorFactory(), Clock.systemUTC(), blobStore, blobIdFactory,
postgresExtension.getPostgresConfiguration(),
new AttachmentIdAssignationStrategy.Default(new StringBackedAttachmentIdFactory()));
Expand All @@ -108,7 +110,7 @@ void beforeAll() throws Exception {
PostgresThreadDAO.Factory threadDAOFactory = new PostgresThreadDAO.Factory(postgresExtension.getExecutorFactory());

eventBus.register(new DeleteMessageListener(blobStore, postgresMailboxMessageDAOFactory, postgresMessageDAOFactory,
attachmentDAOFactory, threadDAOFactory, eventBus));
attachmentDAOFactory, threadDAOFactory,eventBus, postgresConfiguration));

mailboxManager = new PostgresMailboxManager(mapperFactory, sessionProvider,
messageParser, new PostgresMessageId.Factory(),
Expand Down