implemented messages attachments

aligned message to the one in storagehub model

checking if forceExecution is needed

log in creazione

removed drop messages

.vscode/launch.json

@@ -0,0 +1,15 @@
+{
+  "version": "0.2.0",
+  "configurations": [
+    {
+      "type": "java",
+      "name": "Debug Tomcat Remote",
+      "request": "attach",
+      "hostName": "localhost",  // Usa l'IP del server
+      "port": 9000,
+      "projectName": "social-networking-library-ws",
+      "preLaunchTask": "Start SSH Tunnel"  // Task per aprire il tunnel
+
+    }
+  ]
+}

.vscode/settings.json

@@ -0,0 +1,4 @@
+{
+    "java.configuration.updateBuildConfiguration": "interactive",
+    "java.compile.nullAnalysis.mode": "automatic"
+}

.vscode/tasks.json

@@ -0,0 +1,28 @@
+{
+  "version": "2.0.0",
+  "tasks": [
+    {
+      "label": "Start SSH Tunnel",
+      "type": "shell",
+      "command": "ssh -L 9000:localhost:8000 socialservice-dev",
+      "isBackground": true,
+      "problemMatcher": [
+        {
+          "pattern": [
+            {
+              "regexp": ".",
+              "file": 1,
+              "location": 2,
+              "message": 3
+            }
+          ],
+          "background": {
+            "activeOnStart": true,
+            "beginsPattern": ".",
+            "endsPattern": "^Connected"
+          }
+        }
+      ]
+    }
+  ]
+}

CHANGELOG.md

@@ -4,90 +4,99 @@
 All notable changes to this project will be documented in this file.
 This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [v2.2.1-SNAPSHOT]
+
+- moved messages from storagehub to cassandra [#27514]
+- import messages from storagehub
+
+## [v2.1.1]
+
+- tests for Mail notification Catalogue Item Rejectsed [#28020]
+
 ## [v2.1.0] - 2024-10-28
 
- - maven-portal-bom 4.0.0-SNAPSHOT
+- maven-portal-bom 4.0.0-SNAPSHOT
- - [StorageHub] downstream components to upgrade in order to work with storagehub 1.5.0 [#27999]
+- [StorageHub] downstream components to upgrade in order to work with storagehub 1.5.0 [#27999]
 
 ## [v2.0.2] - 2024-10-14
 
- - Mail notification Catalogue Item Rejected [#28020]
+- Mail notification Catalogue Item Rejected [#28020]
 
 ## [v2.0.1] - 2024-04-22
 
- - Null pointer exception getting notifications preferences fixed [#27218]
+- Null pointer exception getting notifications preferences fixed [#27218]
- - Removed noisy logs [#27286]
+- Removed noisy logs [#27286]
 
 ## [v2.0.0] - 2023-12-04
 
- - Support for Cassandra 4.1.3 using DataStax java driver
+- Support for Cassandra 4.1.3 using DataStax java driver
 
 ## [v1.17.0] - 2022-05-13
 
- - Added support for Catalogue notifications
+- Added support for Catalogue notifications
- - Ported to git
+- Ported to git
 
 ## [v1.16.1] - 2018-03-07
 
- - Added JsInterop DTOs (ClientPost, ClientAttachment and JSON) for supporting new IPC Client side in NewsFeed
+- Added JsInterop DTOs (ClientPost, ClientAttachment and JSON) for supporting new IPC Client side in NewsFeed
 
 ## [v1.16.0] - 2017-05-25
 
- - Added feature for hashtags in comments
+- Added feature for hashtags in comments
 
 ## [v1.15.0] - 2017-01-25
 
- - Added support for job completion notifications
+- Added support for job completion notifications
- - Improved exceptions handling
+- Improved exceptions handling
 
 ## [v1.14.0] - 2016-09-29
 
- - Upgraded astyanax dependency to 2.0.2
+- Upgraded astyanax dependency to 2.0.2
- - Removed support for Document Workflow notification
+- Removed support for Document Workflow notification
 
 ## [v1.13.0] - 2016-09-05
 
- - Method to close connections pool to Apache Cassandra added
+- Method to close connections pool to Apache Cassandra added
- - Added methods to retrieve recent user's commented and liked posts
+- Added methods to retrieve recent user's commented and liked posts
- - Added enum class ShowUserStatisticAction
+- Added enum class ShowUserStatisticAction
 
 ## [v1.10.0] - 2016-01-22
 
- - Multi attachments to posts [#1982]
+- Multi attachments to posts [#1982]
 
 ## [v1.9.0] - 2016-01-15
 
- - For user statistics fast retrieval [#1663]
+- For user statistics fast retrieval [#1663]
- - Updated the way we instanciate keyspace, now it is more efficient [#1493]
+- Updated the way we instanciate keyspace, now it is more efficient [#1493]
- - Updated methods for editing comments [#246]
+- Updated methods for editing comments [#246]
 
 ## [v1.8.0] - 2015-07-03
 
- - Added feature for invites and hashtags
+- Added feature for invites and hashtags
 
 ## [v1.6.0] - 2014-04-08
 
- - Added feature for post retrieval by range
+- Added feature for post retrieval by range
- - Added feature for notifications retrieval by range
+- Added feature for notifications retrieval by range
- - Added feature for unlike
+- Added feature for unlike
 
 ## [v1.5.0] - 2014-03-05
 
- - Added feature for post alert notifications
+- Added feature for post alert notifications
 
 ## [v1.3.0] - 2013-07-08
 
- - Added feature for calendar notifications
+- Added feature for calendar notifications
- - Added feature for URL encoding decoding js base 6
+- Added feature for URL encoding decoding js base 6
 
 ## [v1.2.0] - 2013-05-29
 
- - Added feature for fine grained notifications
+- Added feature for fine grained notifications
 
 ## [v1.1.0] - 2013-04-19
 
- - Added feature for people taggings (mentions)
+- Added feature for people taggings (mentions)
- - Fix for method getAllPortalPrivacyLevelFeeds() not recongnizing application posts
+- Fix for method getAllPortalPrivacyLevelFeeds() not recongnizing application posts
 
 ## [v1.0.0] - 2013-01-11
 

messages.cql

@@ -0,0 +1,81 @@
+CREATE TABLE messages_received (
+    recipient_id text,
+    message_id uuid,
+    body text,
+    deleted boolean,
+    from_id text,
+    from_name text,
+    opened boolean,
+    read boolean,
+    subject text,
+    timestamp timestamp,
+    with_attachments boolean,
+    addresses list<text>,
+    PRIMARY KEY (recipient_id, message_id)
+) WITH CLUSTERING ORDER BY (message_id ASC)
+    AND additional_write_policy = '99p'
+    AND bloom_filter_fp_chance = 0.01
+    AND caching = {'keys': 'ALL', 'rows_per_partition': 'NONE'}
+    AND cdc = false
+    AND comment = ''
+    AND compaction = {'class': 'org.apache.cassandra.db.compaction.SizeTieredCompactionStrategy', 'max_threshold': '32', 'min_threshold': '4'}
+    AND compression = {'chunk_length_in_kb': '16', 'class': 'org.apache.cassandra.io.compress.LZ4Compressor'}
+    AND memtable = 'default'
+    AND crc_check_chance = 1.0
+    AND default_time_to_live = 0
+    AND extensions = {}
+    AND gc_grace_seconds = 864000
+    AND max_index_interval = 2048
+    AND memtable_flush_period_in_ms = 0
+    AND min_index_interval = 128
+    AND read_repair = 'BLOCKING'
+    AND speculative_retry = '99p';
+
+CREATE INDEX message_received_deleted_idx ON messages_received (deleted);
+
+CREATE INDEX message_received_id_idx ON messages_received (message_id);
+
+CREATE INDEX message_received_read_idx ON messages_received (read);
+
+CREATE INDEX message_received_timestamp_idx ON messages_received (timestamp);
+
+
+CREATE TABLE messages_sent (
+    from_id text,
+    message_id uuid,
+    body text,
+    deleted boolean,
+    from_name text,
+    opened boolean,
+    read boolean,
+    subject text,
+    timestamp timestamp,
+    with_attachments boolean,
+    addresses list<text>,
+    PRIMARY KEY (from_id, message_id)
+) WITH CLUSTERING ORDER BY (message_id ASC)
+    AND additional_write_policy = '99p'
+    AND bloom_filter_fp_chance = 0.01
+    AND caching = {'keys': 'ALL', 'rows_per_partition': 'NONE'}
+    AND cdc = false
+    AND comment = ''
+    AND compaction = {'class': 'org.apache.cassandra.db.compaction.SizeTieredCompactionStrategy', 'max_threshold': '32', 'min_threshold': '4'}
+    AND compression = {'chunk_length_in_kb': '16', 'class': 'org.apache.cassandra.io.compress.LZ4Compressor'}
+    AND memtable = 'default'
+    AND crc_check_chance = 1.0
+    AND default_time_to_live = 0
+    AND extensions = {}
+    AND gc_grace_seconds = 864000
+    AND max_index_interval = 2048
+    AND memtable_flush_period_in_ms = 0
+    AND min_index_interval = 128
+    AND read_repair = 'BLOCKING'
+    AND speculative_retry = '99p';
+
+CREATE INDEX message_sent_deleted_idx ON messages_sent (deleted);
+
+CREATE INDEX message_sent_id_idx ON messages_sent (message_id);
+
+CREATE INDEX message_sent_read_idx ON messages_sent (read);
+
+CREATE INDEX message_sent_timestamp_idx ON messages_sent (timestamp);

pom.xml

@@ -11,7 +11,7 @@
 
 	<groupId>org.gcube.portal</groupId>
 	<artifactId>social-networking-library</artifactId>
-	<version>2.1.0</version>
+	<version>2.2.1-SNAPSHOT</version>
 	<name>gCube Social Networking Library</name>
 	<description>
 		The gCube Social Networking Library is the 'bridge' between your gCube Applications and the social networking facilities.
@@ -30,7 +30,9 @@
 		<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
 		<cassandra.driver.oss.version>4.13.0</cassandra.driver.oss.version>
 		<logback.version>1.2.3</logback.version>
+		<maven.javadoc.skip>true</maven.javadoc.skip>
 	</properties>
+
 	<dependencyManagement>
 		<dependencies>
 			<dependency>
@@ -64,7 +66,7 @@
 		<dependency>
 			<groupId>org.gcube.resources.discovery</groupId>
 			<artifactId>ic-client</artifactId>
-			<version>1.0.4</version>
+			<version>[1.0.5, 1.1.0-SNAPSHOT]</version>
 			<scope>provided</scope>
 		</dependency>
 		<dependency>

src/main/java/org/gcube/portal/databook/server/CassandraClusterConnection.java

@@ -2,8 +2,11 @@ package org.gcube.portal.databook.server;
 
 import java.net.InetSocketAddress;
 import java.time.Duration;
+import java.util.Arrays;
 import java.util.List;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import com.datastax.oss.driver.api.core.CqlSession;
 import com.datastax.oss.driver.api.core.CqlSessionBuilder;
@@ -11,15 +14,15 @@ import com.datastax.oss.driver.api.core.config.DefaultDriverOption;
 import com.datastax.oss.driver.api.core.config.DriverConfigLoader;
 import com.datastax.oss.driver.api.core.cql.ResultSet;
 import com.datastax.oss.driver.api.core.metadata.Metadata;
+import com.datastax.oss.driver.api.core.metadata.schema.ClusteringOrder;
 import com.datastax.oss.driver.api.core.metadata.schema.KeyspaceMetadata;
 import com.datastax.oss.driver.api.core.type.DataTypes;
 import com.datastax.oss.driver.api.querybuilder.SchemaBuilder;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 /**
  * @author Massimiliano Assante ISTI-CNR
  * @author Ahmed Ibrahim ISTI-CNR
+ * @author Alfredo Oliviero ISTI-CNR
  *
  */
 public class CassandraClusterConnection {
@@ -36,23 +39,45 @@ public class CassandraClusterConnection {
 	private static String keyspaceName;
 	private CqlSession myKeyspaceSession;
 
+	public static CassandraClusterConnection setCustomConfig(String config_host, String config_datacenterName,
+			String config_keyspaceName) throws Exception {
+		return setCustomConfig(config_host, config_datacenterName, config_keyspaceName, false, false);
+	}
+
+	public static CassandraClusterConnection setCustomConfig(String config_host, String config_datacenterName,
+			String config_keyspaceName, boolean dropSchema, boolean forceCreateNew) throws Exception {
+		RunningCluster cluster = RunningCluster.setCustomizedInstance(config_host, config_datacenterName,
+				config_keyspaceName);
+
+		// host = cluster.getHost();
+		hosts = cluster.getHosts();
+		datacenterName = cluster.getDatacenterName();
+		keyspaceName = cluster.getKeyspaceName();
+
+		_log.info("set custom config for CassandraClusterConnection. Hosts " + hosts + ",  datacenterName"
+				+ datacenterName + ", CassandraClusterConnection" + keyspaceName);
+
+		return new CassandraClusterConnection(dropSchema, forceCreateNew);
+	}
+
 	/**
-	 *
 	 * @param dropSchema set true if you want do drop the current and set up new one
 	 *                   the connection to cassandra cluster
+	 * @param forceCreateNew
+	 * @throws Exception
 	 */
-	protected CassandraClusterConnection(boolean dropSchema) throws Exception {
+	public CassandraClusterConnection(boolean dropSchema, boolean forceCreateNew) throws Exception {
 		if (hosts == null || datacenterName == null || keyspaceName == null) {
 			RunningCluster cluster = RunningCluster.getInstance(null);
 
-			//host = cluster.getHost();
+			// host = cluster.getHost();
 			hosts = cluster.getHosts();
 			datacenterName = cluster.getDatacenterName();
 			keyspaceName = cluster.getKeyspaceName();
 		}
 
 		_log.info(keyspaceName + " KeySpace SetUp ...");
-		SetUpKeySpaces(dropSchema);
+		SetUpKeySpaces(dropSchema, forceCreateNew);
 		myKeyspaceSession = connect(keyspaceName);
 		_log.info("CONNECTED! using KeySpace: " + keyspaceName);
 	}
@@ -62,38 +87,61 @@ public class CassandraClusterConnection {
 	 * @param dropSchema set true if you want to drop the current and set up new one
 	 *                   the connection to cassandra cluster
 	 */
-	protected CassandraClusterConnection(boolean dropSchema, String infrastructureName) throws Exception {
+	protected CassandraClusterConnection(boolean dropSchema, boolean forceCreateNew, String infrastructureName)
+			throws Exception {
 		if (hosts == null || datacenterName == null || keyspaceName == null) {
 			RunningCluster cluster = RunningCluster.getInstance(infrastructureName);
-			//host = cluster.getHost();
+			// host = cluster.getHost();
 			hosts = cluster.getHosts();
 			datacenterName = cluster.getDatacenterName();
 			keyspaceName = cluster.getKeyspaceName();
 		}
 		_log.info(keyspaceName + " KeySpace SetUp ...");
-		SetUpKeySpaces(dropSchema);
+		SetUpKeySpaces(dropSchema, forceCreateNew);
 		myKeyspaceSession = connect(keyspaceName);
 		_log.info("CONNECTED! using KeySpace: " + keyspaceName);
 	}
 
-	public CqlSession getKeyspaceSession(){
+	public CqlSession getKeyspaceSession() {
-		if (myKeyspaceSession.isClosed()){
+		if (myKeyspaceSession.isClosed()) {
 			myKeyspaceSession = connect(keyspaceName);
 		}
 		return myKeyspaceSession;
 	}
 
+
 	/**
 	 * @param dropSchema  set true if you want to drop the current and set up new one
 	 *                   the connection to cassandra cluster
+	 * @param forceExecution
 	 */
-	public void SetUpKeySpaces(boolean dropSchema) {
+	public void SetUpKeySpaces(boolean dropSchema, boolean forceExecution) {
 		boolean createNew = false;
 		boolean found = false;
 		CqlSession session = connect();
+
+		Metadata metadata = session.getMetadata();
+		_log.warn("checking if forceExecution is needed ");
+
+		for (String table : Arrays.asList(
+				Schema.TABLE_MESSAGES_SENT,
+				Schema.TABLE_MESSAGES_RECEIVED,
+				Schema.TABLE_MESSAGES_ATTACHMENTS)) {
+			if (forceExecution) {
+				break;
+			}
+			forceExecution = !metadata.getKeyspace(keyspaceName)
+					.flatMap(ks -> ks.getTable(table))
+					.isPresent();
+
+			if (!forceExecution) {
+				_log.warn("missing table {}, forcingExecution of createSchema to update the schema", keyspaceName);
+			}
+		}
+
 		Metadata metaData = session.getMetadata();
 		for (KeyspaceMetadata meta : metaData.getKeyspaces().values()) {
-			if (meta.getName().toString().equals(keyspaceName)){
+			if (meta.getName().toString().equals(keyspaceName)) {
 				found = true;
 				break;
 			}
@@ -115,7 +163,7 @@ public class CassandraClusterConnection {
 				createNew = true;
 			}
 
-			if (!found || createNew) {
+			if (!found || createNew || forceExecution) {
 				_log.info("Keyspace does not exist, triggering schema creation ... ");
 				int replicationFactor = 2;
 				createKeyspace(keyspaceName, replicationFactor);
@@ -141,6 +189,7 @@ public class CassandraClusterConnection {
 		_log.info("[OK] Connected to Cassandra Cluster");
 		return cqlSession;
 	}
+
 	private static CqlSession connect(String KEYSPACE_NAME) {
 		CqlSession cqlSession = configBuilder(CqlSession.builder())
 				.addContactPoints(hosts)
@@ -152,31 +201,34 @@ public class CassandraClusterConnection {
 	}
 
 	public static void closeSession(CqlSession session) {
-		if (session != null) session.close();
+		if (session != null)
+			session.close();
 		_log.info("[OK]Session is now closed");
 	}
 
-	public void closeConnection(){
+	public void closeConnection() {
-		if(!myKeyspaceSession.isClosed()){
+		if (!myKeyspaceSession.isClosed()) {
-			try{
+			try {
 				_log.info("Closing connection");
 				closeSession(myKeyspaceSession);
 				_log.info("Connection closed!");
-			}catch(Exception e){
+			} catch (Exception e) {
 				_log.error("Unable to close connection", e);
 			}
 		}
 	}
 
-	private static CqlSessionBuilder configBuilder(CqlSessionBuilder cqlSessionBuilder){
+	private static CqlSessionBuilder configBuilder(CqlSessionBuilder cqlSessionBuilder) {
 		return cqlSessionBuilder
 				.withConfigLoader(DriverConfigLoader.programmaticBuilder()
-						// Resolves the timeout query 'SELECT * FROM system_schema.tables' timed out after PT2S
+						// Resolves the timeout query 'SELECT * FROM system_schema.tables' timed out
+						// after PT2S
 						.withDuration(DefaultDriverOption.METADATA_SCHEMA_REQUEST_TIMEOUT, Duration.ofMillis(240000))
 						.withDuration(DefaultDriverOption.CONNECTION_INIT_QUERY_TIMEOUT, Duration.ofMillis(240000))
 						.withDuration(DefaultDriverOption.REQUEST_TIMEOUT, Duration.ofMillis(240000))
 						.build());
 	}
+
 	private static void createKeyspace(String keyspaceName, int replicationFactor) {
 		try (CqlSession cqlSession = configBuilder(CqlSession.builder())
 				.addContactPoints(hosts)
@@ -192,7 +244,7 @@ public class CassandraClusterConnection {
 		}
 	}
 
-	private static ResultSet dropKeyspace(){
+	private static ResultSet dropKeyspace() {
 		ResultSet toreturn;
 		try (CqlSession cqlSession = configBuilder(CqlSession.builder())
 				.addContactPoints(hosts)
@@ -204,7 +256,71 @@ public class CassandraClusterConnection {
 		}
 		return toreturn;
 	}
-	private void createTables(){
+
+	private void dropMessagesTables(CqlSession cqlSession) {
+
+		_log.info("Dropping all message tables and indexes");
+
+		// try {
+		cqlSession.execute(
+				SchemaBuilder.dropMaterializedView(Schema.TABLE_MESSAGES_RECEIVED_ACTIVE).ifExists().build());
+		cqlSession.execute(
+				SchemaBuilder.dropMaterializedView(Schema.TABLE_MESSAGES_RECEIVED_READ).ifExists().build());
+		cqlSession
+				.execute(SchemaBuilder.dropMaterializedView(Schema.TABLE_MESSAGES_SENT_ACTIVE).ifExists().build());
+		cqlSession.execute(SchemaBuilder.dropMaterializedView(Schema.TABLE_MESSAGES_SENT_READ).ifExists().build());
+		// } catch (Exception e) {
+		// e.printStackTrace();
+		// }
+
+		// try {
+		cqlSession.execute(SchemaBuilder.dropIndex(Schema.IDX_MESSAGES_SENT_ID).ifExists().build());
+		cqlSession.execute(SchemaBuilder.dropIndex(Schema.IDX_MESSAGES_SENT_READ).ifExists().build());
+		cqlSession.execute(SchemaBuilder.dropIndex(Schema.IDX_MESSAGES_SENT_DELETED).ifExists().build());
+		cqlSession.execute(SchemaBuilder.dropIndex(Schema.IDX_MESSAGES_SENT_CREATION_TIME).ifExists().build());
+
+		cqlSession.execute(SchemaBuilder.dropIndex(Schema.IDX_MESSAGES_RECEIVED_ID).ifExists().build());
+		cqlSession.execute(SchemaBuilder.dropIndex(Schema.IDX_MESSAGES_RECEIVED_READ).ifExists().build());
+		cqlSession.execute(SchemaBuilder.dropIndex(Schema.IDX_MESSAGES_RECEIVED_DELETED).ifExists().build());
+		cqlSession.execute(SchemaBuilder.dropIndex(Schema.IDX_MESSAGES_RECEIVED_CREATION_TIME).ifExists().build());
+		// } catch (Exception e) {
+		// e.printStackTrace();
+		// }
+
+		// try {
+		cqlSession.execute(SchemaBuilder.dropTable(Schema.TABLE_MESSAGES_SENT).ifExists().build());
+		// } catch (Exception e) {
+		// e.printStackTrace();
+		// }
+
+		// try {
+		cqlSession.execute(SchemaBuilder.dropTable(Schema.TABLE_MESSAGES_RECEIVED).ifExists().build());
+		// } catch (Exception e) {
+		// e.printStackTrace();
+		// }
+	}
+
+	private void createMessagesTables(CqlSession cqlSession) {
+
+		// dropMessagesTables(cqlSession);
+
+		// try {
+		createTableMessagesSent(cqlSession);
+		createTableMessagesReceived(cqlSession);
+		createTableMessagesAttachments(cqlSession);
+
+		// } catch (Exception e) {
+		// e.printStackTrace();
+		// }
+
+		// materialized views sono disattivate
+		// createViewMessagesActiveSent(cqlSession);
+		// createViewMessagesActiveReceived(cqlSession);
+		// createViewMessagesUnreadReceived(cqlSession);
+		// createViewMessagesUnreadSent(cqlSession);
+	}
+
+	private void createTables() {
 		try (CqlSession cqlSession = configBuilder(CqlSession.builder())
 				.addContactPoints(hosts)
 				.withLocalDatacenter(datacenterName)
@@ -230,9 +346,15 @@ public class CassandraClusterConnection {
 			createTableNotifications(cqlSession);
 			createTablePosts(cqlSession);
 
+			createMessagesTables(cqlSession);
+
 			closeSession(cqlSession);
+		} catch (Exception e) {
+			_log.error("error creating the schema. {}", e);
+			e.printStackTrace();
 		}
 	}
+
 	private void createTableUSERNotificationsPreferences(CqlSession cqlSession) {
 		cqlSession.execute(SchemaBuilder.createTable("UserNotificationsPreferences")
 				.ifNotExists()
@@ -242,51 +364,56 @@ public class CassandraClusterConnection {
 				.withCompactStorage()
 				.build());
 
-		_log.info("+ Table '{}' has been created (if needed).", "USERNotificationsPreferences");
+		_log.info("+ the Table '{}' has been created (if needed).", "USERNotificationsPreferences");
 	}
+
 	private void createTableUSERNotifications(CqlSession cqlSession) {
 		cqlSession.execute(SchemaBuilder.createTable("UserNotifications")
 				.ifNotExists()
 				.withPartitionKey("userid", DataTypes.TEXT)
-				.withPartitionKey("timestamp", DataTypes.TIMESTAMP)
+				.withPartitionKey(Schema.TIMESTAMP, DataTypes.TIMESTAMP)
 				.withColumn("notid", DataTypes.UUID)
 				.withCompactStorage()
 				.build());
 
 		_log.info("+ Table '{}' has been created (if needed).", "USERNotifications");
 	}
+
 	private void createTableVRETimeline(CqlSession cqlSession) {
 		cqlSession.execute(SchemaBuilder.createTable("VRETimeline")
 				.ifNotExists()
 				.withPartitionKey("vreid", DataTypes.TEXT)
-				.withPartitionKey("timestamp", DataTypes.TIMESTAMP)
+				.withPartitionKey(Schema.TIMESTAMP, DataTypes.TIMESTAMP)
 				.withColumn("postid", DataTypes.UUID)
 				.withCompactStorage()
 				.build());
 
 		_log.info("+ Table '{}' has been created (if needed).", "VRETimeline");
 	}
+
 	private void createTableAppTimeline(CqlSession cqlSession) {
 		cqlSession.execute(SchemaBuilder.createTable("AppTimeline")
 				.ifNotExists()
 				.withPartitionKey("appid", DataTypes.TEXT)
-				.withPartitionKey("timestamp", DataTypes.TIMESTAMP)
+				.withPartitionKey(Schema.TIMESTAMP, DataTypes.TIMESTAMP)
 				.withColumn("postid", DataTypes.UUID)
 				.withCompactStorage()
 				.build());
 
 		_log.info("+ Table '{}' has been created (if needed).", "AppTimeline");
 	}
+
 	private void createTableUSERTimeline(CqlSession cqlSession) {
 		cqlSession.execute(SchemaBuilder.createTable("UserTimeline")
 				.ifNotExists()
 				.withPartitionKey("userid", DataTypes.TEXT)
-				.withPartitionKey("timestamp", DataTypes.TIMESTAMP)
+				.withPartitionKey(Schema.TIMESTAMP, DataTypes.TIMESTAMP)
 				.withColumn("postid", DataTypes.UUID)
 				.withCompactStorage()
 				.build());
 		_log.info("+ Table '{}' has been created (if needed).", "USERTimeline");
 	}
+
 	private void createTableHashtaggedPosts(CqlSession cqlSession) {
 		cqlSession.execute(SchemaBuilder.createTable("HashtaggedPosts")
 				.ifNotExists()
@@ -298,6 +425,7 @@ public class CassandraClusterConnection {
 
 		_log.info("+ Table '{}' has been created (if needed).", "HashtaggedPosts");
 	}
+
 	private void createTableHashtaggedComments(CqlSession cqlSession) {
 		cqlSession.execute(SchemaBuilder.createTable("HashtaggedComments")
 				.ifNotExists()
@@ -309,6 +437,7 @@ public class CassandraClusterConnection {
 
 		_log.info("+ Table '{}' has been created (if needed).", "HashtaggedComments");
 	}
+
 	private void createTableHashtagsCounter(CqlSession cqlSession) {
 		cqlSession.execute(SchemaBuilder.createTable("HashtagsCounter")
 				.ifNotExists()
@@ -320,16 +449,18 @@ public class CassandraClusterConnection {
 
 		_log.info("+ Table '{}' has been created (if needed).", "HashtagsCounter");
 	}
+
 	private void createTableUSERNotificationsUnread(CqlSession cqlSession) {
 		cqlSession.execute(SchemaBuilder.createTable("UserUnreadNotifications")
 				.ifNotExists()
 				.withPartitionKey("userid", DataTypes.TEXT)
-				.withPartitionKey("timestamp", DataTypes.TIMESTAMP)
+				.withPartitionKey(Schema.TIMESTAMP, DataTypes.TIMESTAMP)
 				.withColumn("notid", DataTypes.UUID)
 				.withCompactStorage()
 				.build());
 		_log.info("+ Table '{}' has been created (if needed).", "USERNotificationsUnread");
 	}
+
 	private void createTableUSERLikes(CqlSession cqlSession) {
 		cqlSession.execute(SchemaBuilder.createTable("UserLikes")
 				.ifNotExists()
@@ -341,6 +472,7 @@ public class CassandraClusterConnection {
 
 		_log.info("+ Table '{}' has been created (if needed).", "USERLikes");
 	}
+
 	private void createTableVREInvites(CqlSession cqlSession) {
 		cqlSession.execute(SchemaBuilder.createTable("VREInvites")
 				.ifNotExists()
@@ -352,6 +484,7 @@ public class CassandraClusterConnection {
 
 		_log.info("+ Table '{}' has been created (if needed).", "VREInvites");
 	}
+
 	private void createTableEMAILInvites(CqlSession cqlSession) {
 		cqlSession.execute(SchemaBuilder.createTable("EmailInvites")
 				.ifNotExists()
@@ -362,6 +495,7 @@ public class CassandraClusterConnection {
 				.build());
 		_log.info("+ Table '{}' has been created (if needed).", "EMAILInvites");
 	}
+
 	private void createTableAttachments(CqlSession cqlSession) {
 		cqlSession.execute(SchemaBuilder.createTable("Attachments")
 				.ifNotExists()
@@ -382,6 +516,7 @@ public class CassandraClusterConnection {
 
 		_log.info("+ Table '{}' has been created (if needed).", "Attachments");
 	}
+
 	private void createTableInvites(CqlSession cqlSession) {
 		cqlSession.execute(SchemaBuilder.createTable("Invites")
 				.ifNotExists()
@@ -391,7 +526,7 @@ public class CassandraClusterConnection {
 				.withColumn("email", DataTypes.TEXT)
 				.withColumn("controlcode", DataTypes.TEXT)
 				.withColumn("status", DataTypes.TEXT)
-				.withColumn("timestamp", DataTypes.TIMESTAMP)
+				.withColumn(Schema.TIMESTAMP, DataTypes.TIMESTAMP)
 				.withColumn("senderfullname", DataTypes.TEXT)
 				.withCompactStorage()
 				.build());
@@ -403,6 +538,7 @@ public class CassandraClusterConnection {
 
 		_log.info("+ Table '{}' has been created (if needed).", "Invites");
 	}
+
 	private void createTableLikes(CqlSession cqlSession) {
 		cqlSession.execute(SchemaBuilder.createTable("Likes")
 				.ifNotExists()
@@ -411,7 +547,7 @@ public class CassandraClusterConnection {
 				.withColumn("fullname", DataTypes.TEXT)
 				.withColumn("thumbnailurl", DataTypes.TEXT)
 				.withColumn("postid", DataTypes.UUID)
-				.withColumn("timestamp", DataTypes.TIMESTAMP)
+				.withColumn(Schema.TIMESTAMP, DataTypes.TIMESTAMP)
 				.withCompactStorage()
 				.build());
 		cqlSession.execute(SchemaBuilder.createIndex("post_likes")
@@ -422,6 +558,7 @@ public class CassandraClusterConnection {
 
 		_log.info("+ Table '{}' has been created (if needed).", "Likes");
 	}
+
 	private void createTableComments(CqlSession cqlSession) {
 		cqlSession.execute(SchemaBuilder.createTable("Comments")
 				.ifNotExists()
@@ -431,7 +568,7 @@ public class CassandraClusterConnection {
 				.withColumn("thumbnailurl", DataTypes.TEXT)
 				.withColumn("comment", DataTypes.TEXT)
 				.withColumn("postid", DataTypes.UUID)
-				.withColumn("timestamp", DataTypes.TIMESTAMP)
+				.withColumn(Schema.TIMESTAMP, DataTypes.TIMESTAMP)
 				.withColumn("isedit", DataTypes.BOOLEAN)
 				.withColumn("lastedittime", DataTypes.TIMESTAMP)
 				.withCompactStorage()
@@ -444,6 +581,7 @@ public class CassandraClusterConnection {
 
 		_log.info("+ Table '{}' has been created (if needed).", "Comments");
 	}
+
 	private void createTableNotifications(CqlSession cqlSession) {
 		cqlSession.execute(SchemaBuilder.createTable("Notifications")
 				.ifNotExists()
@@ -467,6 +605,7 @@ public class CassandraClusterConnection {
 				.build());
 		_log.info("+ Table '{}' has been created (if needed).", "Notifications");
 	}
+
 	private void createTablePosts(CqlSession cqlSession) {
 		cqlSession.execute(SchemaBuilder.createTable("Posts")
 				.ifNotExists()
@@ -477,7 +616,7 @@ public class CassandraClusterConnection {
 				.withColumn("likesno", DataTypes.BIGINT)
 				.withColumn("thumbnailurl", DataTypes.TEXT)
 				.withColumn("linkdescription", DataTypes.TEXT)
-				.withColumn("timestamp", DataTypes.TIMESTAMP)
+				.withColumn(Schema.TIMESTAMP, DataTypes.TIMESTAMP)
 				.withColumn("uri", DataTypes.TEXT)
 				.withColumn("isapplicationpost", DataTypes.BOOLEAN)
 				.withColumn("entityid", DataTypes.TEXT)
@@ -500,4 +639,110 @@ public class CassandraClusterConnection {
 		_log.info("+ Table '{}' has been created (if needed).", "Posts");
 	}
 
+	// Method to create the messages_sent table
+	private void createTableMessagesSent(CqlSession cqlSession) {
+		cqlSession.execute(SchemaBuilder.createTable(Schema.TABLE_MESSAGES_SENT)
+				.ifNotExists()
+				.withPartitionKey(Schema.FROM_ID, DataTypes.TEXT) // Partition key on sender ID
+				.withClusteringColumn(Schema.CREATION_TIME, DataTypes.TIMESTAMP) // Clustering by timestamp for ordering
+				.withClusteringColumn(Schema.MESSAGE_ID, DataTypes.UUID) // Clustering by message ID for uniqueness
+				.withColumn(Schema.FROM_NAME, DataTypes.TEXT) // Sender's name
+				.withColumn(Schema.RECIPIENTS, DataTypes.listOf(DataTypes.TEXT)) // List of recipients
+				.withColumn(Schema.SUBJECT, DataTypes.TEXT) // Message subject
+				.withColumn(Schema.BODY, DataTypes.TEXT) // Message body
+				.withColumn(Schema.WITH_ATTACH, DataTypes.BOOLEAN) // Boolean flag for attachments
+				.withColumn(Schema.ISREAD, DataTypes.BOOLEAN) // Boolean flag for read status
+				.withColumn(Schema.ISOPENED, DataTypes.BOOLEAN) // Boolean flag for opened status
+				.withColumn(Schema.ISDELETED, DataTypes.BOOLEAN) // Boolean flag for deleted status
+				.withClusteringOrder(Schema.CREATION_TIME, ClusteringOrder.DESC) // Descending order by timestamp
+				.build());
+
+		_log.info("+ Table '{}' has been created (if needed).", Schema.TABLE_MESSAGES_SENT);
+
+		// Remove unnecessary indexes: no need for message_id or timestamp indexes
+		// Retain only the index for deleted and read messages
+		cqlSession.execute(SchemaBuilder.createIndex(Schema.IDX_MESSAGES_SENT_DELETED)
+				.ifNotExists()
+				.onTable(Schema.TABLE_MESSAGES_SENT)
+				.andColumn(Schema.ISDELETED)
+				.build());
+
+		_log.info("+ Index '{}' has been created (if needed).", Schema.IDX_MESSAGES_SENT_DELETED);
+
+		cqlSession.execute(SchemaBuilder.createIndex(Schema.IDX_MESSAGES_SENT_READ)
+				.ifNotExists()
+				.onTable(Schema.TABLE_MESSAGES_SENT)
+				.andColumn(Schema.ISREAD)
+				.build());
+
+		_log.info("+ Index '{}' has been created (if needed).", Schema.IDX_MESSAGES_SENT_READ);
+
+		cqlSession.execute(SchemaBuilder.createIndex(Schema.IDX_MESSAGES_SENT_ID)
+				.ifNotExists()
+				.onTable(Schema.TABLE_MESSAGES_SENT)
+				.andColumn(Schema.MESSAGE_ID)
+				.build());
+
+		_log.info("+ Index '{}' has been created (if needed).", Schema.IDX_MESSAGES_SENT_ID);
+	}
+
+	private void createTableMessagesAttachments(CqlSession cqlSession) {
+
+		// cqlSession.execute(SchemaBuilder.dropTable(Schema.TABLE_MESSAGES_ATTACHMENTS).ifExists().build());
+
+		cqlSession.execute(SchemaBuilder.createTable(Schema.TABLE_MESSAGES_ATTACHMENTS)
+				.ifNotExists()
+				.withPartitionKey(Schema.MESSAGE_ID, DataTypes.UUID) // Partition key on MESSAGE ID
+				.withClusteringColumn(Schema.ATTACH_ID, DataTypes.UUID) // Clustering by message ID for uniqueness
+
+				.withColumn(Schema.URI, DataTypes.TEXT) // File URL
+				.withColumn(Schema.NAME, DataTypes.TEXT) // File Name
+				.withColumn(Schema.DESCRIPTION, DataTypes.TEXT) // File Description
+				.withColumn(Schema.URI_THUMBNAIL, DataTypes.TEXT) // File Thumb URI
+				.withColumn(Schema.MIME_TYPE, DataTypes.TEXT) // Boolean flag for attachments
+				.build());
+	}
+
+	// Method to create the messages_received table
+	private void createTableMessagesReceived(CqlSession cqlSession) {
+		cqlSession.execute(SchemaBuilder.createTable(Schema.TABLE_MESSAGES_RECEIVED)
+				.ifNotExists()
+				.withPartitionKey(Schema.RECIPIENT_ID, DataTypes.TEXT) // Partition key on recipient ID
+				.withClusteringColumn(Schema.CREATION_TIME, DataTypes.TIMESTAMP) // Clustering by timestamp for ordering
+				.withClusteringColumn(Schema.MESSAGE_ID, DataTypes.UUID) // Clustering by message ID for uniqueness
+				.withColumn(Schema.FROM_ID, DataTypes.TEXT) // Sender's ID
+				.withColumn(Schema.FROM_NAME, DataTypes.TEXT) // Sender's name
+				.withColumn(Schema.RECIPIENTS, DataTypes.listOf(DataTypes.TEXT)) // List of recipients
+				.withColumn(Schema.SUBJECT, DataTypes.TEXT) // Message subject
+				.withColumn(Schema.BODY, DataTypes.TEXT) // Message body
+				.withColumn(Schema.WITH_ATTACH, DataTypes.BOOLEAN) // Boolean flag for attachments
+				.withColumn(Schema.ISREAD, DataTypes.BOOLEAN) // Boolean flag for read status
+				.withColumn(Schema.ISOPENED, DataTypes.BOOLEAN) // Boolean flag for opened status
+				.withColumn(Schema.ISDELETED, DataTypes.BOOLEAN) // Boolean flag for deleted status
+				.withClusteringOrder(Schema.CREATION_TIME, ClusteringOrder.DESC) // Descending order by timestamp
+				.build());
+
+		// Remove unnecessary indexes: no need for message_id or timestamp indexes
+		// Retain only the index for deleted and read messages
+		cqlSession.execute(SchemaBuilder.createIndex(Schema.IDX_MESSAGES_RECEIVED_DELETED)
+				.ifNotExists()
+				.onTable(Schema.TABLE_MESSAGES_RECEIVED)
+				.andColumn(Schema.ISDELETED)
+				.build());
+
+		cqlSession.execute(SchemaBuilder.createIndex(Schema.IDX_MESSAGES_RECEIVED_READ)
+				.ifNotExists()
+				.onTable(Schema.TABLE_MESSAGES_RECEIVED)
+				.andColumn(Schema.ISREAD)
+				.build());
+
+		cqlSession.execute(SchemaBuilder.createIndex(Schema.IDX_MESSAGES_RECEIVED_ID)
+				.ifNotExists()
+				.onTable(Schema.TABLE_MESSAGES_RECEIVED)
+				.andColumn(Schema.MESSAGE_ID)
+				.build());
+
+		_log.info("+ Table '{}' has been created (if needed).", "messages_received");
+	}
+
 }

src/main/java/org/gcube/portal/databook/server/DatabookCassandraTest.java

@@ -1,9 +1,6 @@
 package org.gcube.portal.databook.server;
-import java.util.UUID;
-
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
-import org.junit.Test;
 
 
 public class DatabookCassandraTest {

src/main/java/org/gcube/portal/databook/server/DatabookStore.java

@@ -1,5 +1,6 @@
 package org.gcube.portal.databook.server;
 
+import java.util.Date;
 import java.util.List;
 import java.util.Map;
 
@@ -12,6 +13,9 @@ import org.gcube.portal.databook.shared.Invite;
 import org.gcube.portal.databook.shared.InviteOperationResult;
 import org.gcube.portal.databook.shared.InviteStatus;
 import org.gcube.portal.databook.shared.Like;
+import org.gcube.portal.databook.shared.Message;
+import org.gcube.portal.databook.shared.MessageReceived;
+import org.gcube.portal.databook.shared.MessageSent;
 import org.gcube.portal.databook.shared.Notification;
 import org.gcube.portal.databook.shared.NotificationChannelType;
 import org.gcube.portal.databook.shared.NotificationType;
@@ -30,12 +34,19 @@ import org.gcube.portal.databook.shared.ex.NotificationIDNotFoundException;
 import org.gcube.portal.databook.shared.ex.NotificationTypeNotFoundException;
 import org.gcube.portal.databook.shared.ex.PrivacyLevelTypeNotFoundException;
 
+import com.datastax.oss.driver.api.core.CqlSession;
+
 /**
  * @author Massimiliano Assante ISTI-CNR
  * @author Costantino Perciante ISTI-CNR
+ * @author Alfredo Oliviero ISTI-CNR
+ * 
  * <class>DatabookStore</class> is the high level interface for querying and adding data to DatabookStore 
  */
 public interface DatabookStore {
+
+
+
 	/**
 	 * userid from requests a friendship to userid to 
 	 * @return true if everything went fine
@@ -669,4 +680,38 @@ public interface DatabookStore {
 	 * close the connection to the underlying database
 	 */
 	void closeConnection();
+
+	public Message sendMessage(String fromId, List<String> addresses, String subject, String body, List<Attachment> attachments);
+	public Message sendMessage(String fromId, List<String> addresses, String subject, String body, List<Attachment> attachments, String messageId);
+
+	public List<MessageSent> getSentMessagesBySender(String fromId, Date timestamp, Integer limit);
+	public List<MessageReceived> getReceivedMessagesByRecipient(String recipientId, Date timestamp, Integer limit);
+
+	// public Message getMessageById(String messageId);
+	public MessageSent getSentMessageById(String fromId, String messageId);
+	public MessageReceived getReceivedMessageById(String messageId, String recipientId);
+
+	public Message deleteMessage(String messageId);
+	public MessageSent deleteMessageSent(String fromId, String messageId);
+	public MessageReceived deleteMessageReceived(String messageId, String recipientId);
+
+	public MessageSent setSentMessageRead(String fromId, String messageId, boolean set_read);
+	public MessageReceived setReceivedMessageRead(String recipientId, String messageId, boolean set_read);
+
+	
+	public Attachment getMessageAttachmentByIdFilename(String messageId, String filename);
+	public List<Attachment> getMessageAttachmentsById(String messageId);
+
+	public List<MessageSent> getSentMessagesBySender(String fromId, String messageId, Date timestamp, Integer limit,
+	Boolean filter_deleted, Boolean filter_read, CqlSession session);
+
+	public List<MessageReceived> getReceivedMessagesByRecipient(String recipientId, String messageId, Date timestamp, Integer limit,
+	Boolean filter_deleted, Boolean filter_read, CqlSession session);
+
+	// public MessageSent saveSentMessage(Message message, CqlSession session);
+	// public MessageReceived saveReceivedMessage(Message message, String recipientId, CqlSession session);
+
+	// public boolean saveAttachmentMessageEntry(String messageId, Attachment toSave);
+
+	// public MessageSent saveNewMessage(Message message, List<Attachment> attachments) ;
 }

src/main/java/org/gcube/portal/databook/server/RunningCluster.java

@@ -12,9 +12,6 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.Properties;
 
-import com.datastax.oss.driver.api.core.CqlSession;
-import com.datastax.oss.driver.api.core.metadata.Metadata;
-import com.datastax.oss.driver.api.core.metadata.schema.KeyspaceMetadata;
 import org.gcube.common.portal.GCubePortalConstants;
 import org.gcube.common.portal.PortalContext;
 import org.gcube.common.resources.gcore.ServiceEndpoint;
@@ -25,6 +22,7 @@ import org.gcube.resources.discovery.client.api.DiscoveryClient;
 import org.gcube.resources.discovery.client.queries.api.SimpleQuery;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+
 /**
  * @author Massimiliano Assante ISTI-CNR
  * @author Ahmed Ibrahim ISTI-CNR
@@ -37,8 +35,13 @@ public class RunningCluster implements Serializable {
 	/**
 	 * logger
 	 */
+
 	private static final Logger _log = LoggerFactory.getLogger(RunningCluster.class);
 
+	private final static String ENV_SOCIALDB_HOST = "SOCIALDB_HOST";
+	private final static String ENV_SOCIALDB_DATACENTER = "SOCIALDB_DATACENTER";
+	private final static String ENV_SOCIALDB_KEYSPACE = "SOCIALDB_KEYSPACE";
+
 	/**
 	 * properties to read
 	 */
@@ -55,6 +58,22 @@ public class RunningCluster implements Serializable {
 	private static final String DEFAULT_CONFIGURATION = "/org/gcube/portal/databook/server/resources/databook.properties";
 
 	private static RunningCluster singleton;
+
+	public static synchronized RunningCluster setCustomizedInstance(String host, String datacenterName,
+			String keyspaceName) {
+		singleton = new RunningCluster(host, datacenterName, keyspaceName);
+
+		String[] params = {
+				singleton.host, singleton.datacenterName, singleton.keyspaceName
+		};
+
+		_log.info(
+				"socialdb will use custom configuration host:{}, datacenter:{}, keyspace: {}",
+				params);
+
+		return singleton;
+	}
+
 	/**
 	 * Host
 	 */
@@ -66,35 +85,86 @@ public class RunningCluster implements Serializable {
 	/**
 	 * Keyspace Name
 	 */
-	private String keyspaceName; //to be modified
+	private String keyspaceName; // to be modified
 
 	/**
 	 * @param infrastructureName could be null
 	 * @return an instance of the RunningCluster
 	 */
-	public static synchronized RunningCluster getInstance(String infrastructureName){
+	public static synchronized RunningCluster getInstance(String infrastructureName) {
 		if (singleton == null) {
-			singleton = new RunningCluster(infrastructureName);
+			RunningCluster fromConfig = runningClusterFromEnvConf();
+			if (fromConfig != null) {
+				String[] params = {
+						fromConfig.host, fromConfig.datacenterName, fromConfig.keyspaceName
+				};
+
+				_log.info(
+						"socialdb will use custom configuration from sys environment. host:{}, datacenter:{}, keyspace: {}",
+						params);
+				singleton = fromConfig;
+			}
+
+		}
+
+		if (singleton == null) {
+			RunningCluster fromIS = new RunningCluster(infrastructureName);
+
+			String[] params = {
+					fromIS.host, fromIS.datacenterName, fromIS.keyspaceName
+			};
+
+			_log.info("socialdb will use custom configuration from IS. host:{}, datacenter:{}, keyspace: {}", params);
+			singleton = fromIS;
+
 		}
 		return singleton;
 	}
+
+	private static RunningCluster runningClusterFromEnvConf() {
+		String socialdb_host = System.getenv(ENV_SOCIALDB_HOST);
+		String socialdb_datacenter = System.getenv(ENV_SOCIALDB_DATACENTER);
+		String socialdb_keyspace = System.getenv(ENV_SOCIALDB_KEYSPACE);
+
+		if (socialdb_host == null) {
+			return null;
+		}
+
+		String[] params = {
+				socialdb_host, socialdb_datacenter, socialdb_keyspace
+		};
+
+		_log.info("socialdb will use custom configuration from sys environment. host:{}, datacenter:{}, keyspace: {}",
+				params);
+
+		return new RunningCluster(socialdb_host, socialdb_datacenter, socialdb_keyspace);
+	}
+
+	private RunningCluster(String host, String datacenterName, String keyspaceName) {
+		this.host = host;
+		this.datacenterName = datacenterName;
+		this.keyspaceName = keyspaceName;
+	}
+
 	/**
 	 * private constructor
 	 */
-	private RunningCluster(String infrastructureName){
+	private RunningCluster(String infrastructureName) {
-		//Query the IS (for the future)
+		// Query the IS (for the future)
-		try{
+		try {
 			List<ServiceEndpoint> resources = getConfigurationFromIS(infrastructureName);
 			if (resources.size() > 1) {
-				_log.error("Too many Runtime Resource having name " + RUNTIME_RESOURCE_NAME +" in this scope ");
+				_log.error("Too many Runtime Resource having name " + RUNTIME_RESOURCE_NAME + " in this scope ");
-				throw new TooManyRunningClustersException("There exist more than 1 Runtime Resource in this scope having name " 
+				throw new TooManyRunningClustersException(
-						+ RUNTIME_RESOURCE_NAME + " and Platform " + PLATFORM_NAME + ". Only one allowed per infrasrtucture.");
+						"There exist more than 1 Runtime Resource in this scope having name "
-			}
+								+ RUNTIME_RESOURCE_NAME + " and Platform " + PLATFORM_NAME
-			else if (resources.size() == 0){
+								+ ". Only one allowed per infrasrtucture.");
-				_log.error("There is no Runtime Resource having name " + RUNTIME_RESOURCE_NAME +" and Platform " + PLATFORM_NAME + " in this scope. Using default configuration properties: " + DEFAULT_CONFIGURATION);
+			} else if (resources.size() == 0) {
+				_log.error("There is no Runtime Resource having name " + RUNTIME_RESOURCE_NAME + " and Platform "
+						+ PLATFORM_NAME + " in this scope. Using default configuration properties: "
+						+ DEFAULT_CONFIGURATION);
 				loadDefaultConfiguration();
-			}
+			} else {
-			else {
 				for (ServiceEndpoint res : resources) {
 					AccessPoint found = res.profile().accessPoints().iterator().next();
 					host = found.address();
@@ -104,11 +174,17 @@ public class RunningCluster implements Serializable {
 			}
 		} catch (Exception e) {
 			e.printStackTrace();
+
+			_log.error("Error getting configuration from IS: {} - {}", e.getMessage());
+			_log.error("Error  {}", e);
+			// throw e;
 		}
 
-		/*host = "10.1.28.55:9042, 10.1.30.142:9042, 10.1.28.100:9042";
+		/*
-		datacenterName = "1";
+		 * host = "10.1.28.55:9042, 10.1.30.142:9042, 10.1.28.100:9042";
-		keyspaceName = "dev_mig_consistent";*/
+		 * datacenterName = "1";
+		 * keyspaceName = "dev_mig_consistent";
+		 */
 	}
 
 	/**
@@ -117,9 +193,9 @@ public class RunningCluster implements Serializable {
 	 * @throws Exception
 	 */
 	private List<ServiceEndpoint> getConfigurationFromIS(String infrastructureName) {
-		_log.debug("getConfigurationFromIS infrastructureName="+infrastructureName );
+		_log.debug("getConfigurationFromIS infrastructureName=" + infrastructureName);
 		String scope = "/";
-		if(infrastructureName != null && !infrastructureName.isEmpty())
+		if (infrastructureName != null && !infrastructureName.isEmpty())
 			scope += infrastructureName;
 		else {
 			scope += readInfrastructureName();
@@ -128,8 +204,8 @@ public class RunningCluster implements Serializable {
 		String currScope = ScopeProvider.instance.get();
 		ScopeProvider.instance.set(scope);
 		SimpleQuery query = queryFor(ServiceEndpoint.class);
-		query.addCondition("$resource/Profile/Name/text() eq '"+ RUNTIME_RESOURCE_NAME +"'");
+		query.addCondition("$resource/Profile/Name/text() eq '" + RUNTIME_RESOURCE_NAME + "'");
-		query.addCondition("$resource/Profile/Platform/Name/text() eq '"+ PLATFORM_NAME +"'");
+		query.addCondition("$resource/Profile/Platform/Name/text() eq '" + PLATFORM_NAME + "'");
 		DiscoveryClient<ServiceEndpoint> client = clientFor(ServiceEndpoint.class);
 		List<ServiceEndpoint> toReturn = client.submit(query);
 		ScopeProvider.instance.set(currScope);
@@ -148,16 +224,16 @@ public class RunningCluster implements Serializable {
 			String propertyfile = sb.toString();
 			File propsFile = new File(propertyfile);
 			FileInputStream fis = new FileInputStream(propsFile);
-			props.load( fis);
+			props.load(fis);
 			return props.getProperty(GCubePortalConstants.INFRASTRUCTURE_NAME);
-		}
+		} catch (IOException e) {
-		catch(IOException e) {
+			_log.error(
-			_log.error("infrastructure.properties file not found under $CATALINA_HOME/conf/ dir, setting default infrastructure Name " + "gcube");
+					"infrastructure.properties file not found under $CATALINA_HOME/conf/ dir, setting default infrastructure Name "
+							+ "gcube");
 			return "gcube";
 		}
 	}
 
-
 	/**
 	 *
 	 */
@@ -173,7 +249,6 @@ public class RunningCluster implements Serializable {
 		}
 	}
 
-
 	public String getHost() {
 		return host;
 	}
@@ -182,12 +257,10 @@ public class RunningCluster implements Serializable {
 		this.host = host;
 	}
 
-
 	public String getKeyspaceName() {
 		return keyspaceName;
 	}
 
-
 	public void setKeyspaceName(String keyspaceName) {
 		this.keyspaceName = keyspaceName;
 	}
@@ -197,14 +270,17 @@ public class RunningCluster implements Serializable {
 		return "RunningCluster [host=" + host + ", datacenterName=" + datacenterName
 				+ ", keyspaceName=" + keyspaceName + "]";
 	}
+
 	/**
 	 *
 	 * @return $CATALINA_HOME
 	 */
 	private static String getCatalinaHome() {
-		return (System.getenv("CATALINA_HOME").endsWith("/") ? System.getenv("CATALINA_HOME") : System.getenv("CATALINA_HOME")+"/");
+		return (System.getenv("CATALINA_HOME").endsWith("/") ? System.getenv("CATALINA_HOME")
+				: System.getenv("CATALINA_HOME") + "/");
 	}
-	public void setDatacenterName(String datacenterName){
+
+	public void setDatacenterName(String datacenterName) {
 		this.datacenterName = datacenterName;
 	}
 
@@ -214,8 +290,8 @@ public class RunningCluster implements Serializable {
 
 	public List<InetSocketAddress> getHosts() {
 		List<InetSocketAddress> hosts = new ArrayList<>();
-		String [] ips = host.split(", ");
+		String[] ips = host.split(", ");
-		for (String ip: ips){
+		for (String ip : ips) {
 			String[] ip_port = ip.split(":");
 			hosts.add(new InetSocketAddress(ip_port[0], Integer.parseInt(ip_port[1])));
 		}

src/main/java/org/gcube/portal/databook/server/Schema.java

@@ -1,8 +1,11 @@
 package org.gcube.portal.databook.server;
 
+import com.datastax.oss.driver.api.core.CqlIdentifier;
+
 /**
  * @author Massimiliano Assante ISTI-CNR
  * @author Ahmed Ibrahim ISTI-CNR
+ * @author Alfredo Oliviero ISTI-CNR
  *
  * @version 2.0.0 October 2023
  *
@@ -73,4 +76,51 @@ public class Schema {
     public static final String COMMENTS_NO = "commentsno"; //big int
     public static final String LINK_TITLE = "linktitle"; //text
 
+    // messages
+
+
+    public static final String TABLE_MESSAGES_SENT = "messages_sent";
+    public static final String TABLE_MESSAGES_RECEIVED = "messages_received";
+    public static final String TABLE_MESSAGES_ATTACHMENTS = "messages_attachments";
+
+    public static final String IDX_MESSAGES_SENT_ID = "message_sent_id_idx";
+    public static final String IDX_MESSAGES_SENT_READ = "message_sent_read_idx";
+    public static final String IDX_MESSAGES_SENT_DELETED = "message_sent_deleted_idx";
+    public static final String IDX_MESSAGES_SENT_CREATION_TIME = "message_sent_creation_time_idx";
+
+
+    public static final String IDX_MESSAGES_RECEIVED_ID = "message_received_id_idx";
+    public static final String IDX_MESSAGES_RECEIVED_READ = "message_received_read_idx";
+    public static final String IDX_MESSAGES_RECEIVED_DELETED = "message_received_deleted_idx";
+    public static final String IDX_MESSAGES_RECEIVED_CREATION_TIME = "message_received_creation_time_idx";
+
+
+
+    public static final String TABLE_MESSAGES_SENT_ACTIVE = "messages_active_sent";
+    public static final String TABLE_MESSAGES_RECEIVED_ACTIVE = "messages_active_received";
+    public static final String TABLE_MESSAGES_SENT_READ = "messages_unread_sent";
+    public static final String TABLE_MESSAGES_RECEIVED_READ = "messages_unread_received";
+
+    
+
+    // public static final String URI = "uri"; //text
+    // public static final String URI_THUMBNAIL = "urithumbnail"; //text
+    // public static final String NAME = "name"; //text
+    // public static final String DESCRIPTION = "description"; //text
+    // public static final String MIME_TYPE = "mimetype";  //text
+
+
+    public static final String FROM_ID = "from_id"; //text
+    public static final String FROM_NAME = "from_name"; //text
+    public static final String CREATION_TIME = "creation_time"; //text
+    public static final String MESSAGE_ID = "message_id"; //text
+    public static final String RECIPIENT_ID = "recipient_id"; //text
+    public static final String RECIPIENTS = "addresses"; //text
+    // public static final String RECIPIENTS = "recipients"; //text
+    public static final String SUBJECT = "subject"; //text
+    public static final String BODY = "body"; //text
+    public static final String WITH_ATTACH = "with_attachments"; //text
+    public static final String ISREAD = "read"; //text
+    public static final String ISOPENED = "opened"; //text
+    public static final String ISDELETED = "deleted"; //text
 }

src/main/java/org/gcube/portal/databook/server/Tester.java

@@ -1,53 +0,0 @@
-package org.gcube.portal.databook.server;
-
-import org.gcube.portal.databook.shared.*;
-import org.gcube.portal.databook.shared.ex.*;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.List;
-
-/**
- * @author Massimiliano Assante ISTI-CNR
- * @author Ahmed Ibrahim ISTI-CNR
- *
- * @version 2.0.0 October 2023
- *
- */
-public class Tester {
-    private static DBCassandraAstyanaxImpl store;
-    private static Logger LOGGER = LoggerFactory.getLogger(Tester.class);
-
-    public Tester() {
-        store = new DBCassandraAstyanaxImpl("gcube"); //set to true if you want to drop the KeySpace and recreate it
-        
-    }
-
-    public static void main(String[] args) throws ColumnNameNotFoundException, PrivacyLevelTypeNotFoundException, FeedIDNotFoundException, FeedTypeNotFoundException {
-        Tester test = new Tester();
-        //test.getComment();
-        test.testFunc();
-        System.exit(0);
-
-    }
-    public  void testFunc() throws ColumnNameNotFoundException, PrivacyLevelTypeNotFoundException, FeedIDNotFoundException, FeedTypeNotFoundException {
-        String postIdToUpdate = "047c601d-2291-4974-9224-d6732b1fbe26";
-        Post read = store.readPost(postIdToUpdate);
-
-        List<Comment> readC = store.getAllCommentByPost("047c601d-2291-4974-9224-d6732b1fbe26");
-        System.out.println(read);
-        readC.forEach(c -> System.out.println(c.getText()));
-    }
-    public void getComment(){
-        String uuid = "820969b2-4632-4197-9fd6-5aafab781faa";
-
-        Comment c;
-        try {
-            c = store.readCommentById(uuid);
-            System.out.println(c);
-        } catch (CommentIDNotFoundException e) {
-            // TODO Auto-generated catch block
-            System.err.println(e.toString());
-        }
-    }
-}

src/main/java/org/gcube/portal/databook/shared/Message.java

@@ -0,0 +1,167 @@
+package org.gcube.portal.databook.shared;
+
+import java.io.Serializable;
+import java.util.Date;
+import java.util.List;
+import java.util.UUID;
+
+/**
+ * 
+ * @author Alfredo Oliviero, ISTI-CNR
+ *
+ */
+@SuppressWarnings("serial")
+public class Message implements Serializable, Comparable<Message> {
+
+	// WORKSPACE
+	// public class MessageUser {
+	// String user_name;
+	// String user_id;
+	// }
+	// "id": "cb3059b7-24ac-4bf0-8e37-525ce85a2c72",
+	// "name": "78b321f9-793b-4474-a440-38c0a38f205f",
+	// "path": "/Home/alfredo.oliviero/InBox/78b321f9-793b-4474-a440-38c0a38f205f",
+	// "parent_id": "847fa2b2-bcfa-48c1-b513-9267eb215c05",
+	// "parent_path": "/Home/alfredo.oliviero/InBox",
+	// "primary_type": "nthl:itemSentRequestSH",
+	// "with_attachments": false,
+	// "sender": {
+	// "user_name": "lucio.lelii",
+	// "user_id": "lucio.lelii"
+	// },
+	// "subject": "prova",
+	// "body": "asdfsadsfsdadfsadfsadf",
+	// "read": true,
+	// "opened": false,
+	// "addresses": [
+	// "alfredo.oliviero"
+	// ],
+	// "creation_time": 1722869475845
+
+	protected String id;
+	// protected String name;
+	// protected String path;
+	// protected String parent_id;
+	// protected String parent_path;
+	// protected String primary_type;
+	protected boolean with_attachments;
+
+	// protected MessageUser sender;
+	protected String user_id;
+	protected String user_name;
+
+	protected List<String> addresses;
+
+	protected String subject;
+	protected String body;
+
+	// protected Date creation_time;
+	protected Date creation_time;
+
+	public UUID getUUID() {
+		return UUID.fromString(getId());
+	}
+
+	public String getId() {
+		return id;
+	}
+
+	public String getKey() {
+		return id;
+	}
+
+	public void setId(String id) {
+		this.id = id;
+	}
+
+	public boolean isWith_attachments() {
+		return with_attachments;
+	}
+
+	public void setWith_attachments(boolean with_attachments) {
+		this.with_attachments = with_attachments;
+	}
+
+	public String getUser_id() {
+		return user_id;
+	}
+
+	public void setUser_id(String userid) {
+		this.user_id = userid;
+	}
+
+	public String getUser_name() {
+		return user_name;
+	}
+
+	public void setUser_name(String user_name) {
+		this.user_name = user_name;
+	}
+
+	public String getSubject() {
+		return subject;
+	}
+
+	public void setSubject(String subject) {
+		this.subject = subject;
+	}
+
+	public String getBody() {
+		return body;
+	}
+
+	public void setBody(String body) {
+		this.body = body;
+	}
+
+
+	public List<String> getAddresses() {
+		return addresses;
+	}
+
+	public void setAddresses(List<String> addresses) {
+		this.addresses = addresses;
+	}
+
+	public Date getCreation_time() {
+		return creation_time;
+	}
+
+	public void setCreation_time(Date timestamp) {
+		this.creation_time = timestamp;
+	}
+
+	public Message() {
+		super();
+	}
+
+	public static String generateUUID(){
+		UUID uuid = UUID.randomUUID();
+		return uuid.toString();
+	}
+
+	public Message(String id, String userid, String user_name, List<String> addresses, String subject, String body,
+			Date timestamp, boolean with_attachments ) {
+		super();
+		this.id = id;
+		this.user_id = userid;
+		this.user_name = user_name;
+		this.addresses = addresses;
+		this.subject = subject;
+		this.body = body;
+		this.creation_time = timestamp;
+		this.with_attachments = with_attachments;
+	}
+
+	public int compareTo(Message toCompare) {
+		return this.getCreation_time().compareTo(toCompare.getCreation_time());
+	}
+
+	@Override
+	public String toString() {
+		return "Message [id=" + id + ", user_id=" + user_id + ", creation_time=" + creation_time
+				+ ", addresses=" + addresses + ", subject=" + subject + ", body="
+				+ body + ", with_attachments=" + with_attachments + "]";
+	}
+}
+

src/main/java/org/gcube/portal/databook/shared/MessageReceived.java

@@ -0,0 +1,72 @@
+package org.gcube.portal.databook.shared;
+
+import java.util.Date;
+import java.util.List;
+
+public class MessageReceived extends Message {
+    protected boolean read;
+    protected boolean opened;
+    protected String recipient_id;
+    protected boolean deleted;
+
+    public boolean isDeleted() {
+        return deleted;
+    }
+
+    public void setDeleted(boolean deleted) {
+        this.deleted = deleted;
+    }
+
+    public String getRecipient_id() {
+        return recipient_id;
+    }
+
+    public void setRecipient_id(String recipientid) {
+        this.recipient_id = recipientid;
+    }
+
+    public MessageReceived(String id, String recipientid, String userid, String user_name, List<String> addresses,
+            String subject, String body,
+            Date timestamp, boolean with_attachments, boolean read, boolean opened, boolean deleted) {
+        super(id, userid, user_name, addresses, subject, body, timestamp, with_attachments);
+        this.recipient_id = recipientid;
+        this.read = read;
+        this.opened = opened;
+        this.deleted = deleted;
+    }
+
+    public MessageReceived(Message message,
+            String recipientid, boolean read, boolean opened, boolean deleted) {
+        super(message.getId(), message.getUser_id(), message.getUser_name(), message.getAddresses(),
+                message.getSubject(), message.getBody(), message.getCreation_time(), message.isWith_attachments() );
+        this.recipient_id = recipientid;
+        this.read = read;
+        this.opened = opened;
+        this.deleted = deleted;
+    }
+
+    @Override
+    public String toString() {
+        return "ReceivedMessage [id=" + id + ", user_id=" + user_id + ", creation_time=" + creation_time
+                + ", addresses=" + addresses + ", subject=" + subject + ", body="
+                + body + ", with_attachments=" + with_attachments +
+                ", recipientid=" + recipient_id + ", read=" + read + ", opened=" + opened + ", deleted=" + deleted + "]";
+    }
+
+    public boolean isRead() {
+        return read;
+    }
+
+    public void setRead(boolean read) {
+        this.read = read;
+    }
+
+    public boolean isOpened() {
+        return opened;
+    }
+
+    public void setOpened(boolean opened) {
+        this.opened = opened;
+    }
+
+}

src/main/java/org/gcube/portal/databook/shared/MessageSent.java

@@ -0,0 +1,62 @@
+package org.gcube.portal.databook.shared;
+
+import java.util.Date;
+import java.util.List;
+
+public class MessageSent extends Message {
+    protected boolean read;
+    protected boolean opened;
+    protected String recipientid;
+
+    protected boolean deleted;
+
+    public boolean isDeleted() {
+        return deleted;
+    }
+
+    public void setDeleted(boolean deleted) {
+        this.deleted = deleted;
+    }
+
+    public MessageSent(String id, String userid, String user_name, List<String> addresses, String subject,
+            String body,
+            Date timestamp, boolean with_attachments, boolean read, boolean opened, boolean deleted) {
+        super(id, userid, user_name, addresses, subject, body, timestamp, with_attachments);
+        this.deleted = deleted;
+        this.read = read;
+        this.opened = opened;
+    }
+
+    public MessageSent(Message message, boolean read, boolean opened, boolean deleted) {
+        super(message.getId(), message.getUser_id(), message.getUser_name(), message.getAddresses(),
+                message.getSubject(), message.getBody(), message.getCreation_time(), message.isWith_attachments());
+        this.deleted = deleted;
+        this.read = read;
+        this.opened = opened;
+    }
+
+    @Override
+    public String toString() {
+        return "SentMessage [id=" + id + ", user_id=" + user_id + ", creation_time=" + creation_time
+                + ", addresses=" + addresses + ", subject=" + subject + ", body="
+                + body + ", with_attachments=" + with_attachments + ", read="
+                + read + ", opened=" + opened + ", deleted=" + deleted + "]";
+    }
+
+    public boolean isRead() {
+        return read;
+    }
+
+    public void setRead(boolean read) {
+        this.read = read;
+    }
+
+    public boolean isOpened() {
+        return opened;
+    }
+
+    public void setOpened(boolean opened) {
+        this.opened = opened;
+    }
+
+}

src/test/java/org/gcube/portal/databook/EmptyTest.java

@@ -0,0 +1,13 @@
+package org.gcube.portal.databook;
+
+import org.junit.Test;
+
+public class EmptyTest {
+
+
+    @Test
+    public void empty() {
+        System.out.println("Testing works");
+        return;
+    }
+}

src/test/java/org/gcube/portal/databook/server/BaseDbTest.java

@@ -0,0 +1,89 @@
+package org.gcube.portal.databook.server;
+
+import static org.junit.Assert.assertNotNull;
+
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * @author Alfredo Oliviero ISTI-CNR
+ **/
+
+public class BaseDbTest {
+    protected DBCassandraAstyanaxImpl store;
+
+    protected static String INFRASTRUCTURE = "gcube";
+
+    // Dockerized cassandra
+    protected final static String HOST = "127.0.0.1:9042";
+    protected final static String DATACENTER = "DC1";
+    protected final static String KEYSPACE = "dev_keyspace_1";
+
+    // dev parameters
+    // protected final static String HOST = "10.1.28.55:9042, 10.1.30.142:9042,
+    // 10.1.28.100:9042";
+    // protected final static String DATACENTER = "d4science_datacenter";
+    // protected final static String KEYSPACE = "dev_keyspace";
+
+    // protected static String host = null;
+    // protected static String datacenterName = null;
+    // protected static String keyspace = null;
+
+    @Before
+    public void setUp() throws Exception {
+        if (CUSTOM_DB)
+            setCustomConfig();
+        store = createStore();
+    }
+
+    // set true to override the configuration from ic-client
+    protected static boolean CUSTOM_DB = false;
+
+    // used when the custom ic-client is not available
+    protected static void setCustomConfig() throws Exception {
+        setCustomConfig(HOST, DATACENTER, KEYSPACE);
+    }
+
+    protected static void setCustomConfig(String host, String datacenterName, String keysSpace) throws Exception {
+        // host = HOST;
+        // datacenterName = DATACENTER;
+        // keyspace = KEYSPACE;
+        CassandraClusterConnection.setCustomConfig(host, datacenterName, keysSpace);
+    }
+
+    public static DBCassandraAstyanaxImpl createStore() throws Exception {
+        return new DBCassandraAstyanaxImpl(INFRASTRUCTURE);
+    }
+
+    public DBCassandraAstyanaxImpl getStore() throws Exception {
+        if (store == null) {
+            store = createStore();
+        }
+        assertNotNull(store);
+        return store;
+    }
+
+    @Test
+    public void testStore() throws Exception{
+        store = getStore();
+        assertNotNull(store);
+    }
+    public CassandraClusterConnection getConnection() throws Exception {
+        return new CassandraClusterConnection(false,  false, null);
+    }
+
+    @Test
+    public void testConnection() throws Exception {
+        System.out.print("starting test createConnection");
+        CassandraClusterConnection connection = getConnection();
+        assertNotNull(connection);
+        System.out.print("connection extablished" + connection);
+    }
+
+    @Test
+    public void empty() {
+        System.out.print("Testing works");
+        return;
+    }
+
+}

src/test/java/org/gcube/portal/databook/server/DbTest.java

@@ -0,0 +1,54 @@
+package org.gcube.portal.databook.server;
+
+import static org.junit.Assert.assertNotNull;
+
+import java.util.List;
+
+import org.junit.Test;
+
+import com.datastax.oss.driver.api.core.CqlSession;
+import com.datastax.oss.driver.api.core.cql.ResultSet;
+import com.datastax.oss.driver.api.core.cql.Row;
+import com.datastax.oss.driver.api.core.cql.SimpleStatement;
+
+/**
+ * @author Alfredo Oliviero ISTI-CNR
+ **/
+
+public class DbTest extends BaseDbTest {
+
+    @Test
+    public void updateSchema() throws Exception {
+        CassandraClusterConnection conn = getConnection();
+        conn.SetUpKeySpaces(false, true);
+    }
+
+    @Test
+    public void cresteStore() throws Exception {
+        System.out.print("starting test cresteStore");
+        CassandraClusterConnection connection = getConnection();
+
+        System.out.print("connection extablished" + connection);
+        DBCassandraAstyanaxImpl store = getStore();
+        assertNotNull(store);
+        System.out.print("store created" + store);
+    }
+
+    @Test
+    public void baseQuery() throws Exception {
+        CqlSession session = getConnection().getKeyspaceSession();
+
+        SimpleStatement test_count = SimpleStatement.builder(
+                "select * from vretimeline limit 1")
+                // .addPositionalValues(set_read, userId, UUID.fromString(messageId))
+                .build();
+
+        ResultSet result = session.execute(test_count);
+        List<Row> all = result.all();
+        org.junit.Assert.assertTrue(all.size() == 1);
+
+        // _log.info("baseTest found {}", all.get(0));
+        System.out.print("baseTest found" + all.get(0));
+    }
+
+}

src/test/java/org/gcube/portal/databook/server/MessagesTest.java

@@ -0,0 +1,685 @@
+package org.gcube.portal.databook.server;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Date;
+import java.util.List;
+import java.util.UUID;
+
+import org.gcube.portal.databook.shared.Attachment;
+import org.gcube.portal.databook.shared.Message;
+import org.gcube.portal.databook.shared.MessageReceived;
+import org.gcube.portal.databook.shared.MessageSent;
+import org.gcube.portal.databook.shared.Notification;
+import org.gcube.portal.databook.shared.ex.NotificationTypeNotFoundException;
+import org.junit.Before;
+import org.junit.Test;
+
+import com.datastax.oss.driver.api.core.CqlSession;
+
+public class MessagesTest extends BaseDbTest{
+
+    // private CassandraClusterConnection connection;
+    // private DBCassandraAstyanaxImpl store;
+    private String testFrom;
+    private String recipient;
+    private String recipient2;
+    private String subject;
+    private String body;
+    private List<String> recipients;
+
+    @Before
+    public void setUp() throws Exception {
+        // connection = DbTest.getConnection();
+        testFrom = "alfredo.oliviero@isti.cnr.it";
+        recipient = "massimiliano.assante@isti.cnr.it";
+        recipient2 = "andrea.rossi@isti.cnr.it";
+        subject = "test subject";
+        body = "test body";
+        recipients = Arrays.asList(recipient, recipient2);
+        store = getStore();
+    }
+
+    @Test
+    public void empty() {
+        System.out.print("Testing works");
+        return;
+    }
+
+    @Test
+    public void testCreateMessage() throws Exception {
+        Message message = store.sendMessage(testFrom, recipients, subject, body, null);
+
+        assertNotNull("Message should not be null after creation", message);
+        assertEquals("User ID does not match", testFrom, message.getUser_id());
+        assertEquals("Recipient does not match", recipient, message.getAddresses().get(0));
+        assertEquals("Subject does not match", subject, message.getSubject());
+        assertEquals("Body does not match", body, message.getBody());
+
+        MessageSent messageSent = store.getSentMessageById(testFrom, message.getId());
+
+        assertNotNull("Sent message should not be null", messageSent);
+        assertEquals("User ID of sent message does not match", testFrom, messageSent.getUser_id());
+        assertEquals("Recipient of sent message does not match", recipient, messageSent.getAddresses().get(0));
+        assertEquals("Subject of sent message does not match", subject, messageSent.getSubject());
+        assertEquals("Body of sent message does not match", body, messageSent.getBody());
+
+        MessageReceived messageReceived = store.getReceivedMessageById(recipient, message.getId());
+
+        assertNotNull("Received message should not be null", messageReceived);
+        assertEquals("User ID of received message does not match", testFrom, messageReceived.getUser_id());
+        assertEquals("Recipient of received message does not match", recipient, messageReceived.getRecipient_id());
+        assertEquals("Subject of received message does not match", subject, messageReceived.getSubject());
+        assertEquals("Body of received message does not match", body, messageReceived.getBody());
+
+        MessageReceived messageReceived2 = store.getReceivedMessageById(recipient2, message.getId());
+
+        assertNotNull("Received2 message should not be null", messageReceived2);
+        assertEquals("User ID of received2 message does not match", testFrom, messageReceived2.getUser_id());
+        assertEquals("Recipient of received2 message does not match", recipient2, messageReceived2.getRecipient_id());
+        assertEquals("Subject of received2 message does not match", subject, messageReceived2.getSubject());
+        assertEquals("Body of received2 message does not match", body, messageReceived2.getBody());
+    }
+
+    @Test
+    public void testMessagesReceivedByRecipient() throws Exception {
+        Message message = store.sendMessage(testFrom, recipients, "testMessagesReceivedByRecipient", body, null);
+        assertNotNull(message);
+
+        List<MessageReceived> receiveds = store.getReceivedMessagesByRecipient(recipient, null, null);
+        MessageReceived received = receiveds.get(0);
+        assertNotNull(received);
+    }
+
+    @Test
+    public void testUnreadMessagesReceivedByRecipient() throws Exception {
+        Message message = store.sendMessage(testFrom, recipients, "testUnreadMessagesReceivedByRecipient", body, null);
+        assertNotNull(message);
+
+        List<MessageReceived> receiveds = store.getReceivedMessagesByRecipient(recipient, null, null, 1, false, false,
+                null);
+        MessageReceived received = receiveds.get(0);
+        assertFalse(received.isRead());
+        assertNotNull(received);
+    }
+
+    @Test
+    public void testMessageReceivedById() throws Exception {
+        // String msg_id = "d78e096e-df88-4783-9e8e-32036158f54b";
+        Message message = store.sendMessage(testFrom, recipients, "testMessageReceivedById", body, null);
+        String msg_id = message.getId();
+
+        MessageReceived received = store.getReceivedMessageById(recipient, msg_id);
+
+        assertNotNull(received);
+        assertEquals(received.getId(), msg_id);
+    }
+
+    @Test
+    public void testSetSentMessageReadUnread() throws Exception {
+        Message message = store.sendMessage(testFrom, recipients, "testSetSentMessageReadUnread", body, null);
+        assertNotNull(message);
+
+        String recipient = recipients.get(0);
+        String recipient1 = recipients.get(1);
+
+        MessageSent messageSent = store.getSentMessageById(testFrom, message.getId());
+        MessageReceived messageReceived = store.getReceivedMessageById(recipient, message.getId());
+        MessageReceived messageReceived1 = store.getReceivedMessageById(recipient1, message.getId());
+
+        assertNotNull(messageSent);
+        assertNotNull(messageReceived);
+        assertNotNull(messageReceived1);
+
+        MessageSent sentMessageUpdate = store.setSentMessageRead(testFrom, message.getId(), true);
+        messageSent = store.getSentMessageById(testFrom, message.getId());
+        messageReceived = store.getReceivedMessageById(recipient, message.getId());
+        messageReceived1 = store.getReceivedMessageById(recipient1, message.getId());
+
+        assertNotNull(sentMessageUpdate);
+        assertNotNull(messageReceived);
+        assertNotNull(messageReceived1);
+
+        assertTrue(messageSent.isRead());
+        assertTrue(sentMessageUpdate.isRead());
+
+        assertFalse(messageReceived.isRead());
+        assertFalse(messageReceived1.isRead());
+
+        sentMessageUpdate = store.setSentMessageRead(testFrom, message.getId(), false);
+        messageSent = store.getSentMessageById(testFrom, message.getId());
+        messageReceived = store.getReceivedMessageById(recipient, message.getId());
+        messageReceived1 = store.getReceivedMessageById(recipient1, message.getId());
+
+        assertNotNull(sentMessageUpdate);
+        assertNotNull(messageReceived);
+        assertNotNull(messageReceived1);
+
+        assertFalse(messageSent.isRead());
+        assertFalse(sentMessageUpdate.isRead());
+
+        assertFalse(messageReceived.isRead());
+        assertFalse(messageReceived1.isRead());
+
+        sentMessageUpdate = store.setSentMessageRead(testFrom, message.getId(), true);
+        messageSent = store.getSentMessageById(testFrom, message.getId());
+        messageReceived = store.getReceivedMessageById(recipient, message.getId());
+        messageReceived1 = store.getReceivedMessageById(recipient1, message.getId());
+
+        assertNotNull(sentMessageUpdate);
+        assertNotNull(messageReceived);
+        assertNotNull(messageReceived1);
+
+        assertTrue(messageSent.isRead());
+        assertTrue(sentMessageUpdate.isRead());
+
+        assertFalse(messageReceived.isRead());
+        assertFalse(messageReceived1.isRead());
+
+    }
+
+    @Test
+    public void testSetReceivedMessageReadUnread() throws Exception {
+        Message message = store.sendMessage(testFrom, recipients, "testSetReceivedMessageReadUnread", body, null);
+        assertNotNull(message);
+
+        String recipient = recipients.get(0);
+        String recipient1 = recipients.get(1);
+
+        MessageSent messageSent = store.getSentMessageById(testFrom, message.getId());
+        MessageReceived messageReceived = store.getReceivedMessageById(recipient, message.getId());
+        MessageReceived messageReceived1 = store.getReceivedMessageById(recipient1, message.getId());
+
+        assertNotNull(messageSent);
+        assertNotNull(messageReceived);
+        assertNotNull(messageReceived1);
+
+        // set read for recipient
+        MessageReceived messageUpdated = store.setReceivedMessageRead(recipient, message.getId(), true);
+
+        messageSent = store.getSentMessageById(testFrom, message.getId());
+        messageReceived = store.getReceivedMessageById(recipient, message.getId());
+        messageReceived1 = store.getReceivedMessageById(recipient1, message.getId());
+
+        assertNotNull(messageUpdated);
+        assertNotNull(messageReceived);
+        assertNotNull(messageReceived1);
+
+        assertTrue(messageReceived.isRead());
+        assertTrue(messageUpdated.isRead());
+
+        assertFalse(messageSent.isRead());
+        assertFalse(messageReceived1.isRead());
+
+        // set not read for recipient
+        messageUpdated = store.setReceivedMessageRead(recipient, message.getId(), false);
+
+        messageSent = store.getSentMessageById(testFrom, message.getId());
+        messageReceived = store.getReceivedMessageById(recipient, message.getId());
+        messageReceived1 = store.getReceivedMessageById(recipient1, message.getId());
+
+        assertNotNull(messageUpdated);
+        assertNotNull(messageReceived);
+        assertNotNull(messageReceived1);
+
+        assertFalse(messageReceived.isRead());
+        assertFalse(messageUpdated.isRead());
+        assertFalse(messageSent.isRead());
+        assertFalse(messageReceived1.isRead());
+
+        // set read for recipient2
+        messageUpdated = store.setReceivedMessageRead(recipient2, message.getId(), true);
+
+        messageSent = store.getSentMessageById(testFrom, message.getId());
+        messageReceived = store.getReceivedMessageById(recipient, message.getId());
+        messageReceived1 = store.getReceivedMessageById(recipient1, message.getId());
+
+        assertNotNull(messageUpdated);
+        assertNotNull(messageReceived);
+        assertNotNull(messageReceived1);
+
+        assertTrue(messageReceived1.isRead());
+        assertTrue(messageUpdated.isRead());
+
+        assertFalse(messageSent.isRead());
+        assertFalse(messageReceived.isRead());
+
+    }
+
+    @Test
+    public void testDeleteSentMessage() throws Exception {
+        Message message = store.sendMessage(testFrom, recipients, "testDeleteSentMessage", body, null);
+        assertNotNull(message);
+
+        String recipient = recipients.get(0);
+        String recipient1 = recipients.get(1);
+
+        MessageSent messageSent = store.getSentMessageById(testFrom, message.getId());
+        MessageReceived messageReceived = store.getReceivedMessageById(recipient, message.getId());
+        MessageReceived messageReceived1 = store.getReceivedMessageById(recipient1, message.getId());
+
+        assertNotNull(messageSent);
+        assertNotNull(messageReceived);
+        assertNotNull(messageReceived1);
+
+        assertFalse(messageSent.isDeleted());
+        assertFalse(messageReceived.isDeleted());
+        assertFalse(messageReceived.isDeleted());
+
+        MessageSent deletedMessage = store.deleteMessageSent(testFrom, message.getId());
+
+        assertNotNull(deletedMessage);
+        assertTrue(deletedMessage.isDeleted());
+
+        messageSent = store.getSentMessageById(testFrom, message.getId());
+
+        messageReceived = store.getReceivedMessageById(recipient, message.getId());
+        messageReceived1 = store.getReceivedMessageById(recipient1, message.getId());
+
+        assertNull(messageSent);
+        assertNotNull(messageReceived);
+        assertNotNull(messageReceived1);
+
+    }
+
+    @Test
+    public void testDeleteReceivedMessage() throws Exception {
+        Message message = store.sendMessage(testFrom, recipients, "testDeleteReceivedMessage", body, null);
+        assertNotNull(message);
+
+        String recipient = recipients.get(0);
+        String recipient1 = recipients.get(1);
+
+        MessageSent messageSent = store.getSentMessageById(testFrom, message.getId());
+        MessageReceived messageReceived = store.getReceivedMessageById(recipient, message.getId());
+        MessageReceived messageReceived1 = store.getReceivedMessageById(recipient1, message.getId());
+
+        assertNotNull(messageSent);
+        assertNotNull(messageReceived);
+        assertNotNull(messageReceived1);
+
+        assertFalse(messageSent.isDeleted());
+        assertFalse(messageReceived.isDeleted());
+        assertFalse(messageReceived.isDeleted());
+
+        MessageReceived deletedMessage = store.deleteMessageReceived(recipient, message.getId());
+
+        assertNotNull(deletedMessage);
+        assertTrue(deletedMessage.isDeleted());
+
+        messageSent = store.getSentMessageById(testFrom, message.getId());
+        messageReceived = store.getReceivedMessageById(recipient, message.getId());
+        messageReceived1 = store.getReceivedMessageById(recipient1, message.getId());
+
+        assertNotNull(messageSent);
+        assertNull(messageReceived);
+        assertNotNull(messageReceived1);
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testGetInvalidIdMessage() {
+        store.getSentMessageById(testFrom, "not-valid-id");
+    }
+
+    @Test
+    public void testGetNonExistentMessage() {
+        MessageSent messageSent = store.getSentMessageById(testFrom, "ffffaaaa-9d90-4cc8-a634-bf158c7cc068");
+        assertNull("Expected null for non-existent message", messageSent);
+    }
+
+    @Test
+    public void testGetNonExistentFrom() {
+        MessageSent messageSent = store.getSentMessageById("aaaa", "ffffaaaa-9d90-4cc8-a634-bf158c7cc068");
+        assertNull("Expected null for non-existent message", messageSent);
+    }
+
+    // @Test
+    // public void testSaveNewMessageWithNullMessage() {
+    // store.saveNewMessage(null, null, null); // Message is null
+    // assertEquals("message cannot be null or empty", exception.getMessage());
+    // }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testSaveNewMessageWithNullUserId() {
+        store.sendMessage(null, recipients, subject, body, null); // User ID is null
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testSaveNewMessageWithNullAddresses() {
+        store.sendMessage(testFrom, null, subject, body, null); // Addresses are null
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testSaveNewMessageWithEmptyAddresses() {
+        store.sendMessage(testFrom, Arrays.asList(), subject, body, null); // Addresses are empty
+    }
+
+    @Test
+    public void testGetSentMessagesOnlyNonDeleted() throws Exception {
+        // Invia un messaggio normale
+        Message message1 = store.sendMessage(testFrom, recipients, subject, body, null);
+
+        // Invia un altro messaggio che verrà cancellato
+        Message message2 = store.sendMessage(testFrom, recipients, subject + " deleted", body, null);
+
+        // Cancella il secondo messaggio
+        store.deleteMessageSent(testFrom, message2.getId());
+
+        // Recupera tutti i messaggi inviati dall'utente
+        List<MessageSent> sentMessages = store.getSentMessagesBySender(testFrom, null, 10);
+
+        // Verifica che non contenga il messaggio cancellato
+        assertNotNull("Sent messages list should not be null", sentMessages);
+        assertFalse("Sent messages list should not contain deleted messages",
+                sentMessages.stream().anyMatch(m -> m.getId().equals(message2.getId())));
+
+        // Verifica che il messaggio non cancellato sia ancora presente
+        assertTrue("Sent messages list should contain non-deleted message",
+                sentMessages.stream().anyMatch(m -> m.getId().equals(message1.getId())));
+    }
+
+    @Test
+    public void testSentMessagesOrderedByTimestampDescending() throws Exception {
+        // Invia tre messaggi con timestamp differenti
+        Message message1 = store.sendMessage(testFrom, recipients, subject + "1", body, null);
+        assertNotNull(message1);
+
+        Thread.sleep(1000); // Attendi 1 secondo per garantire un timestamp diverso
+        Message message2 = store.sendMessage(testFrom, recipients, subject + "2", body, null);
+        assertNotNull(message2);
+
+        Thread.sleep(1000); // Attendi un altro secondo
+        Message message3 = store.sendMessage(testFrom, recipients, subject + "3", body, null);
+        assertNotNull(message3);
+
+
+        // Recupera tutti i messaggi inviati dall'utente
+        List<MessageSent> sentMessages = store.getSentMessagesBySender(testFrom, null, 10);
+
+        // Verifica che la lista non sia nulla e contenga almeno tre messaggi
+        assertNotNull("Sent messages list should not be null", sentMessages);
+        assertTrue("Sent messages list should contain at least 3 messages", sentMessages.size() >= 3);
+
+        // Verifica che i messaggi siano ordinati in modo decrescente in base al
+
+        // timestamp
+        Date timestamp1 = sentMessages.get(0).getCreation_time();
+        Date timestamp2 = sentMessages.get(1).getCreation_time();
+        Date timestamp3 = sentMessages.get(2).getCreation_time();
+
+        assertTrue("First message should be newer than second", timestamp1.after(timestamp2));
+        assertTrue("Second message should be newer than third", timestamp2.after(timestamp3));
+    }
+
+    @Test
+    public void testReceivedMessagesOrderedByTimestampDescending() throws Exception {
+        // Invia tre messaggi con timestamp differenti
+        Message message1 = store.sendMessage(testFrom, recipients, subject + "1", body, null);
+        assertNotNull(message1);
+
+        Thread.sleep(1000); // Attendi 1 secondo per garantire un timestamp diverso
+        Message message2 = store.sendMessage(testFrom, recipients, subject + "2", body, null);
+        assertNotNull(message2);
+
+        Thread.sleep(1000); // Attendi un altro secondo
+        Message message3 = store.sendMessage(testFrom, recipients, subject + "3", body, null);
+        assertNotNull(message3);
+
+
+        // Recupera tutti i messaggi ricevuti dal destinatario
+        List<MessageReceived> receivedMessages = store.getReceivedMessagesByRecipient(recipient, null, 10);
+
+        // Verifica che la lista non sia nulla e contenga almeno tre messaggi
+        assertNotNull("Received messages list should not be null", receivedMessages);
+        assertTrue("Received messages list should contain at least 3 messages", receivedMessages.size() >= 3);
+
+        // Verifica che i messaggi siano ordinati in modo decrescente in base al
+        // timestamp
+        Date timestamp1 = receivedMessages.get(0).getCreation_time();
+        Date timestamp2 = receivedMessages.get(1).getCreation_time();
+        Date timestamp3 = receivedMessages.get(2).getCreation_time();
+
+        assertTrue("First message should be newer than second", timestamp1.after(timestamp2));
+        assertTrue("Second message should be newer than third", timestamp2.after(timestamp3));
+    }
+
+    @Test
+    public void testUnreadMessagesOnlyReturned() throws Exception {
+        // Invia un messaggio normale
+        Message message1 = store.sendMessage(testFrom, recipients, subject + "1", body, null);
+        assertNotNull(message1);
+
+
+        // Invia un altro messaggio che verrà marcato come letto
+        Message message2 = store.sendMessage(testFrom, recipients, subject + "2", body, null);
+        store.setReceivedMessageRead(recipient, message2.getId(), true); // Marca come letto
+
+        // Recupera solo i messaggi non letti
+        List<MessageReceived> unreadMessages = store.getReceivedMessagesByRecipient(recipient, null, null, 10, false,
+                false, null);
+
+        assertNotNull("Unread messages list should not be null", unreadMessages);
+        assertTrue("Unread messages list should contain only unread messages",
+                unreadMessages.stream().allMatch(m -> !m.isRead()));
+    }
+
+    @Test
+    public void testMessagesWithLimitAndOrder() throws Exception {
+        // Invia tre messaggi
+        Message message1 = store.sendMessage(testFrom, recipients, subject + "1", body, null);
+        assertNotNull(message1);
+
+        Thread.sleep(1000); // Attendi 1 secondo
+        Message message2 = store.sendMessage(testFrom, recipients, subject + "2", body, null);
+        assertNotNull(message2);
+
+        Thread.sleep(1000);
+        Message message3 = store.sendMessage(testFrom, recipients, subject + "3", body, null);
+        assertNotNull(message3);
+
+
+        // Recupera i primi due messaggi inviati, ordinati in modo decrescente per
+        // timestamp
+        List<MessageSent> sentMessages = store.getSentMessagesBySender(testFrom, null, 2);
+
+        assertNotNull("Sent messages list should not be null", sentMessages);
+        assertEquals("Sent messages list should contain 2 messages", 2, sentMessages.size());
+
+        // Verifica che i messaggi siano ordinati per timestamp decrescente
+        Date timestamp1 = sentMessages.get(0).getCreation_time();
+        Date timestamp2 = sentMessages.get(1).getCreation_time();
+        assertTrue("First message should be newer than second", timestamp1.after(timestamp2));
+    }
+
+    @Test
+    public void testRetrieveMessagesWithLimit() throws Exception {
+        int count = 4;
+        for (int i = 0; i < count; i++) {
+            store.sendMessage(testFrom, recipients, subject + " " + i, body, null);
+        }
+        ;
+
+        // Recupera solo 2 messaggi
+        List<MessageSent> sentMessages = store.getSentMessagesBySender(testFrom, null, 2);
+        assertNotNull("Sent messages list should not be null", sentMessages);
+        assertEquals("Only 2 messages should be returned", 2, sentMessages.size());
+    }
+
+    @Test
+    public void testGetSentMessagesBySenderWithTimestampFilter() throws Exception {
+        // Invia tre messaggi con timestamp differenti
+        Message message1 = store.sendMessage(testFrom, recipients, subject + "1", body, null);
+        Thread.sleep(1000); // Attendi 1 secondo per garantire un timestamp diverso
+        Message message2 = store.sendMessage(testFrom, recipients, subject + "2", body, null);
+        Thread.sleep(1000); // Attendi un altro secondo
+        Message message3 = store.sendMessage(testFrom, recipients, subject + "3", body, null);
+
+        // Usa il timestamp del secondo messaggio per il test
+        Date filterTimestamp = message2.getCreation_time();
+
+        // Recupera tutti i messaggi con timestamp <= filterTimestamp
+        List<MessageSent> sentMessages = store.getSentMessagesBySender(testFrom, filterTimestamp, 10);
+
+        // Verifica che la lista non sia nulla
+        assertNotNull("Sent messages list should not be null", sentMessages);
+
+        // Verifica che i messaggi abbiano timestamp <= filterTimestamp
+        for (MessageSent message : sentMessages) {
+            assertTrue("Message timestamp should be less than or equal to the filter timestamp",
+                    message.getCreation_time().compareTo(filterTimestamp) <= 0);
+        }
+
+        // Verifica che il terzo messaggio (con timestamp maggiore) non sia nella lista
+        assertFalse("Message with timestamp greater than filterTimestamp should not be returned",
+                sentMessages.stream().anyMatch(m -> m.getId().equals(message3.getId())));
+    }
+
+    @Test
+    public void testGetReceivedMessagesByRecipientWithTimestampFilter() throws Exception {
+        // Invia tre messaggi con timestamp differenti
+        Message message1 = store.sendMessage(testFrom, recipients, subject + "1", body, null);
+        Thread.sleep(1000); // Attendi 1 secondo per garantire un timestamp diverso
+        Message message2 = store.sendMessage(testFrom, recipients, subject + "2", body, null);
+        Thread.sleep(1000); // Attendi un altro secondo
+        Message message3 = store.sendMessage(testFrom, recipients, subject + "3", body, null);
+
+        // Usa il timestamp del secondo messaggio per il test
+        Date filterTimestamp = message2.getCreation_time();
+
+        // Recupera tutti i messaggi ricevuti con timestamp <= filterTimestamp
+        List<MessageReceived> receivedMessages = store.getReceivedMessagesByRecipient(recipient, filterTimestamp, 10);
+
+        // Verifica che la lista non sia nulla
+        assertNotNull("Received messages list should not be null", receivedMessages);
+
+        // Verifica che i messaggi abbiano timestamp <= filterTimestamp
+        for (MessageReceived message : receivedMessages) {
+            assertTrue("Message timestamp should be less than or equal to the filter timestamp",
+                    message.getCreation_time().compareTo(filterTimestamp) <= 0);
+        }
+
+        // Verifica che il terzo messaggio (con timestamp maggiore) non sia nella lista
+        assertFalse("Message with timestamp greater than filterTimestamp should not be returned",
+                receivedMessages.stream().anyMatch(m -> m.getId().equals(message3.getId())));
+    }
+
+    @Test
+    public void testGetEmptyAttachmentByMessageId() throws Exception {
+        Message msg = store.sendMessage(testFrom, recipients, "testGetAttachmentByMessageId", body, null);
+        List<Attachment> attachs = store.getMessageAttachmentsById(msg.getId());
+        assertEquals("attachs list should be empty", attachs.size(), 0);
+    }
+
+    @Test
+    public void testGetSingleAttachmentByMessageId() throws Exception {
+        Attachment attach = new Attachment();
+        String attachId = UUID.randomUUID().toString(); // Generate UUID for the message
+        String filename = "file name";
+        String description = "file description";
+        String uri = "uri";
+        String thumb_url = "thumb url";
+
+        attach.setId(attachId);
+        attach.setName(filename);
+        attach.setDescription(description);
+        attach.setUri(uri);
+        attach.setThumbnailURL(thumb_url);
+
+        List<Attachment> attachments = Arrays.asList(attach);
+        Message msg = store.sendMessage(testFrom, recipients, "testGetAttachmentByMessageId", body, attachments);
+
+        List<Attachment> attachs = store.getMessageAttachmentsById(msg.getId());
+        assertEquals("attachs list should have 1 element", 1, attachs.size());
+
+        Attachment obtained_attach = attachs.get(0);
+
+        assertEquals("attachs list should have 1 element", attach.getId(), obtained_attach.getId());
+        checkEqualAttachments(attach, obtained_attach);
+    }
+
+    @Test
+    public void testMultiSingleAttachmentByMessageId() throws Exception {
+
+        String filename = "file name";
+        String description = "file description";
+        String uri = "uri";
+        String thumb_url = "thumb url";
+
+        List<Attachment> attachments = new ArrayList<Attachment>();
+
+        int count = 10;
+
+        for (int i = 0; i < count; i++) {
+            Attachment attach = new Attachment();
+            String attachId = UUID.randomUUID().toString(); // Generate UUID for the message
+            attach.setId(attachId);
+            attach.setName(filename + " " + i);
+            attach.setDescription(description + " " + i);
+            attach.setUri(uri + " " + i);
+            attach.setThumbnailURL(thumb_url + " " + i);
+
+            attachments.add(attach);
+        }
+
+        Message msg = store.sendMessage(testFrom, recipients, "testGetAttachmentByMessageId", body, attachments);
+
+        String msg_id = msg.getId();
+        List<Attachment> attachs_by_message_id = store.getMessageAttachmentsById(msg_id);
+        assertEquals("attachs list should have  element", count, attachs_by_message_id.size());
+
+        for (Attachment attach : attachments) {
+            String attach_id = attach.getId();
+
+            Attachment by_message_id = attachs_by_message_id.stream().filter(c -> c.getId().equals(attach_id))
+                    .findFirst().orElse(null);
+
+            assertNotNull("cannot obtain attach ", attach_id + "by msg id " + msg_id);
+            checkEqualAttachments(attach, by_message_id);
+
+            Attachment by_attach_id = store.getMessageAttachmentByIdFilename(msg_id, attach_id);
+            assertNotNull("cannot obtain attach by id", attach_id);
+            checkEqualAttachments(attach, by_attach_id);
+        }
+    }
+
+    private boolean checkEqualAttachments(Attachment attach, Attachment obtained_attach) {
+        assertEquals("wrong attachId", attach.getId(), obtained_attach.getId());
+        assertEquals("wrong filename", attach.getName(), obtained_attach.getName());
+        assertEquals("wrong description", attach.getDescription(), obtained_attach.getDescription());
+        assertEquals("wrong uri", attach.getUri(), obtained_attach.getUri());
+        assertEquals("wrong thumb uri", attach.getThumbnailURL(), obtained_attach.getThumbnailURL());
+
+        return true;
+    }
+
+    @Test
+    public void populate_messages() throws NotificationTypeNotFoundException {
+        CqlSession session = store.getSession();
+        Date timestamp = null;
+        // Integer limit = 100000;
+         Integer limit = 10;
+
+        List<String> destinatari = Arrays.asList("alfredo.oliviero", "massimiliano.assante", "andrea.rossi");
+        List<Notification> notifications = store.getAllNotifications(timestamp, limit, session);
+
+        for (Notification n : notifications) {
+            // protected MessageUser sender;
+            String sender = n.getSenderid();
+            Message m = new Message(Message.generateUUID(), n.getSenderid(), n.getSenderFullName(), destinatari,
+                    n.getSubjectid().toString(), n.getDescription(), n.getTime(), true);
+            Attachment attach = new Attachment(Message.generateUUID(), n.getUri(), m.getId(), n.getDescription(),
+                    n.getSenderThumbnail(), "image/png");
+            store.sendMessage(m, Arrays.asList(attach), session);
+            if (!destinatari.contains(sender))
+                destinatari.set(0, sender);
+            System.out.println("created message " + m);
+        }
+    }
+
+}

src/test/java/org/gcube/portal/databook/server/NotificationsTest.java

@@ -0,0 +1,115 @@
+package org.gcube.portal.databook.server;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+
+import org.gcube.portal.databook.shared.NotificationChannelType;
+import org.gcube.portal.databook.shared.NotificationType;
+import org.gcube.portal.databook.shared.ex.NotificationChannelTypeNotFoundException;
+import org.gcube.portal.databook.shared.ex.NotificationTypeNotFoundException;
+import org.junit.Test;
+
+import com.datastax.oss.driver.api.core.CqlSession;
+import com.datastax.oss.driver.api.core.cql.ResultSet;
+import com.datastax.oss.driver.api.core.cql.Row;
+import com.datastax.oss.driver.api.core.cql.SimpleStatement;
+import com.datastax.oss.driver.api.querybuilder.QueryBuilder;
+import com.datastax.oss.driver.api.querybuilder.delete.Delete;
+import com.datastax.oss.driver.api.querybuilder.select.Select;
+
+public class NotificationsTest extends BaseDbTest {
+
+    protected String userId = "alfredo.oliviero";
+    // protected String userId = "alina.sirbu";
+
+    @Test
+    public void baseQuery() throws Exception {
+        CqlSession session = getConnection().getKeyspaceSession();
+
+        SimpleStatement test_count = SimpleStatement.builder(
+                "select * from vretimeline limit 1")
+                // .addPositionalValues(set_read, userId, UUID.fromString(messageId))
+                .build();
+
+        ResultSet result = session.execute(test_count);
+        List<Row> all = result.all();
+        org.junit.Assert.assertTrue(all.size() == 1);
+
+        // _log.info("baseTest found {}", all.get(0));
+        System.out.print("baseTest found" + all.get(0));
+    }
+
+    @Test
+    public void preferencesNotificatonsConfig()
+            throws NotificationTypeNotFoundException, NotificationChannelTypeNotFoundException, Exception {
+        Map<NotificationType, NotificationChannelType[]> preferences = getStore()
+                .getUserNotificationPreferences(userId);
+
+        for (NotificationType type : preferences.keySet()) {
+            System.out.println("## preferences type " + type);
+            for (NotificationChannelType channelType : preferences.get(type)) {
+                System.out.println("## -- " + channelType);
+            }
+        }
+    }
+
+    @Test
+    public void prefNotificatonChannels()
+            throws NotificationChannelTypeNotFoundException, NotificationTypeNotFoundException, Exception {
+        List<NotificationChannelType> channels = getStore().getUserNotificationChannels(userId,
+                NotificationType.CAT_ITEM_PUBLISHED);
+        for (NotificationChannelType channel : channels) {
+            System.out.println("## -- " + channel);
+        }
+    }
+
+    @Test
+    public void getCatPreferencesQuery() throws Exception {
+        Select select = QueryBuilder.selectFrom(Schema.USER_NOTIFICATIONS_PREFERENCES)
+                .all()
+                .whereColumn("userid").isEqualTo(QueryBuilder.literal(userId))
+                .whereColumn("type").in(
+                        // QueryBuilder.literal("WP_ITEM_NEW"),
+                        QueryBuilder.literal("CAT_ITEM_DELETE"),
+                        QueryBuilder.literal("CAT_ITEM_PUBLISHED"),
+                        QueryBuilder.literal("CAT_ITEM_REJECTED"),
+                        QueryBuilder.literal("CAT_ITEM_SUBMITTED"),
+                        QueryBuilder.literal("CAT_ITEM_UPDATED"));
+
+        SimpleStatement statement = select.build();
+        CqlSession session = getConnection().getKeyspaceSession();
+        ResultSet resultSet = session.execute(statement);
+
+        for (Row row : resultSet) {
+            // Supponendo che la tua tabella abbia colonne "userid", "type", e altre
+            String userId = row.getString("userid");
+            String type = row.getString("type");
+
+            // Stampa i valori delle colonne
+            System.out.println("UserId: " + userId + ", Type: " + type);
+        }
+    }
+
+    @Test
+    public void deleteCatPreferencesQuery() throws Exception {
+        Delete delete = QueryBuilder.deleteFrom(Schema.USER_NOTIFICATIONS_PREFERENCES)
+                .whereColumn("userid").isEqualTo(QueryBuilder.literal(userId))
+                .whereColumn("type").in(
+                        QueryBuilder.literal("CAT_ITEM_DELETE"),
+                        QueryBuilder.literal("CAT_ITEM_PUBLISHED"),
+                        QueryBuilder.literal("CAT_ITEM_REJECTED"),
+                        QueryBuilder.literal("CAT_ITEM_SUBMITTED"),
+                        QueryBuilder.literal("CAT_ITEM_UPDATED"));
+
+        SimpleStatement statement = delete.build();
+        CqlSession session = getConnection().getKeyspaceSession();
+        ResultSet resultSet = session.execute(statement);
+
+        for (Row row : resultSet) {
+            // Stampa i valori delle colonne
+            System.out.println(row.getFormattedContents());
+        }
+    }
+
+}

src/test/resources/.gitignore

@@ -0,0 +1,2 @@
+ic_data
+ic-client.properties