feat: implement annotation-driven event observer framework

Author: Twiineenock

hiero-enterprise-base/src/main/java/org/hiero/base/implementation/TopicRepositoryImpl.java

@@ -31,6 +31,14 @@ public TopicRepositoryImpl(@NonNull final MirrorNodeClient mirrorNodeClient) {
     return mirrorNodeClient.queryTopicMessages(topicId);
   }
 
+  @Override
+  public @NonNull Page<TopicMessage> getMessages(@NonNull TopicId topicId, @NonNull java.time.Instant after)
+      throws HieroException {
+    Objects.requireNonNull(topicId, "topicId must not be null");
+    Objects.requireNonNull(after, "after must not be null");
+    return mirrorNodeClient.queryTopicMessages(topicId, after);
+  }
+
   @Override
   public @NonNull Optional<TopicMessage> getMessageBySequenceNumber(
       TopicId topicId, long sequenceNumber) throws HieroException {

hiero-enterprise-base/src/main/java/org/hiero/base/implementation/TransactionRepositoryImpl.java

@@ -29,6 +29,14 @@ public Page<TransactionInfo> findByAccount(@NonNull final AccountId accountId)
     return this.mirrorNodeClient.queryTransactionsByAccount(accountId);
   }
 
+  @Override
+  public @NonNull Page<TransactionInfo> findByAccount(
+      @NonNull AccountId accountId, @NonNull java.time.Instant after) throws HieroException {
+    Objects.requireNonNull(accountId, "accountId must not be null");
+    Objects.requireNonNull(after, "after must not be null");
+    return this.mirrorNodeClient.queryTransactionsByAccount(accountId, after);
+  }
+
   @Override
   public @NonNull Page<TransactionInfo> findByAccountAndType(
       @NonNull AccountId accountId, @NonNull TransactionType type) throws HieroException {

hiero-enterprise-base/src/main/java/org/hiero/base/mirrornode/MirrorNodeClient.java

@@ -175,6 +175,18 @@ default Optional<Nft> queryNftsByAccountAndTokenIdAndSerial(
   @NonNull Page<TransactionInfo> queryTransactionsByAccount(@NonNull AccountId accountId)
       throws HieroException;
 
+  /**
+   * Queries transactions for a specific account after a certain timestamp.
+   *
+   * @param accountId the account ID to query transactions for
+   * @param after the timestamp to start querying from (exclusive)
+   * @return a page of transaction information
+   * @throws HieroException if an error occurs during the query
+   */
+  @NonNull
+  Page<TransactionInfo> queryTransactionsByAccount(@NonNull AccountId accountId, @NonNull java.time.Instant after)
+      throws HieroException;
+
   /**
    * Queries all transactions for a specific account and transaction type.
    *
@@ -396,6 +408,18 @@ default Optional<Topic> queryTopicById(String topicId) throws HieroException {
    */
   @NonNull Page<TopicMessage> queryTopicMessages(TopicId topicId) throws HieroException;
 
+  /**
+   * Return TopicMessages for given topicId after a certain timestamp.
+   *
+   * @param topicId id of the topic
+   * @param after the timestamp to start querying from (exclusive)
+   * @return Page of TopicMessage
+   * @throws HieroException if the search fails
+   */
+  @NonNull
+  Page<TopicMessage> queryTopicMessages(@NonNull TopicId topicId, @NonNull java.time.Instant after)
+      throws HieroException;
+
   /**
    * Return TopicMessages for given topicId.
    *

hiero-enterprise-base/src/main/java/org/hiero/base/mirrornode/TopicRepository.java

@@ -1,6 +1,7 @@
 package org.hiero.base.mirrornode;
 
 import com.hedera.hashgraph.sdk.TopicId;
+import java.time.Instant;
 import java.util.Objects;
 import java.util.Optional;
 import org.hiero.base.HieroException;
@@ -57,7 +58,34 @@ default Page<TopicMessage> getMessages(String topicId) throws HieroException {
     Objects.requireNonNull(topicId, "topicId must not be null");
     return getMessages(TopicId.fromString(topicId));
   }
-  ;
+
+  /**
+   * Return TopicMessages for given topicId after a certain timestamp.
+   *
+   * @param topicId id of the topic
+   * @param after the timestamp to start querying from (exclusive)
+   * @return Page of TopicMessage
+   * @throws HieroException if the search fails
+   */
+  @NonNull
+  Page<TopicMessage> getMessages(@NonNull TopicId topicId, @NonNull Instant after)
+      throws HieroException;
+
+  /**
+   * Return TopicMessages for given topicId after a certain timestamp.
+   *
+   * @param topicId id of the topic
+   * @param after the timestamp to start querying from (exclusive)
+   * @return Page of TopicMessage
+   * @throws HieroException if the search fails
+   */
+  @NonNull
+  default Page<TopicMessage> getMessages(@NonNull String topicId, @NonNull Instant after)
+      throws HieroException {
+    Objects.requireNonNull(topicId, "topicId must not be null");
+    Objects.requireNonNull(after, "after must not be null");
+    return getMessages(TopicId.fromString(topicId), after);
+  }
 
   /**
    * Return TopicMessage for given topicId.

hiero-enterprise-base/src/main/java/org/hiero/base/mirrornode/TransactionRepository.java

@@ -1,6 +1,7 @@
 package org.hiero.base.mirrornode;
 
 import com.hedera.hashgraph.sdk.AccountId;
+import java.time.Instant;
 import java.util.Objects;
 import java.util.Optional;
 import org.hiero.base.HieroException;
@@ -38,6 +39,34 @@ default Page<TransactionInfo> findByAccount(@NonNull String accountId) throws Hi
     return findByAccount(AccountId.fromString(accountId));
   }
 
+  /**
+   * Find all transactions associated with a specific account after a certain timestamp.
+   *
+   * @param accountId id of the account
+   * @param after the timestamp to start querying from (exclusive)
+   * @return page of transactions
+   * @throws HieroException if the search fails
+   */
+  @NonNull
+  Page<TransactionInfo> findByAccount(@NonNull AccountId accountId, @NonNull Instant after)
+      throws HieroException;
+
+  /**
+   * Find all transactions associated with a specific account after a certain timestamp.
+   *
+   * @param accountId id of the account as a string
+   * @param after the timestamp to start querying from (exclusive)
+   * @return page of transactions
+   * @throws HieroException if the search fails
+   */
+  @NonNull
+  default Page<TransactionInfo> findByAccount(@NonNull String accountId, @NonNull Instant after)
+      throws HieroException {
+    Objects.requireNonNull(accountId, "accountId must not be null");
+    Objects.requireNonNull(after, "after must not be null");
+    return findByAccount(AccountId.fromString(accountId), after);
+  }
+
   /**
    * Find all transactions associated with a specific account and has specific transaction type.
    *

hiero-enterprise-base/src/main/java/org/hiero/base/observer/AbstractPollingObserver.java

@@ -0,0 +1,92 @@
+package org.hiero.base.observer;
+
+import java.time.Duration;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.jspecify.annotations.NonNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Base implementation for observers that poll the Mirror Node at regular intervals.
+ *
+ * <p>This class handles the lifecycle of the background polling task, including
+ * starting, stopping, and handling unexpected errors during polling.
+ *
+ * @param <T> The type of event being observed.
+ */
+public abstract class AbstractPollingObserver<T> {
+
+    private static final Logger log = LoggerFactory.getLogger(AbstractPollingObserver.class);
+
+    private final ScheduledExecutorService executorService;
+    private final Duration pollingInterval;
+    private final EventObserver<T> listener;
+    private final AtomicBoolean running = new AtomicBoolean(false);
+
+    /**
+     * Creates a new polling observer.
+     *
+     * @param pollingInterval How often to check for new events.
+     * @param listener The callback to trigger when an event is found.
+     */
+    protected AbstractPollingObserver(@NonNull Duration pollingInterval, @NonNull EventObserver<T> listener) {
+        this.pollingInterval = pollingInterval;
+        this.listener = listener;
+        this.executorService = Executors.newSingleThreadScheduledExecutor(r -> {
+            Thread thread = new Thread(r, "hiero-observer-" + getClass().getSimpleName());
+            thread.setDaemon(true);
+            return thread;
+        });
+    }
+
+    /**
+     * Starts the background polling task.
+     */
+    public void start() {
+        if (running.compareAndSet(false, true)) {
+            log.info("Starting Hiero observer: {}", getClass().getSimpleName());
+            executorService.scheduleAtFixedRate(this::pollSafe, 0, pollingInterval.toMillis(), TimeUnit.MILLISECONDS);
+        }
+    }
+
+    /**
+     * Stops the background polling task.
+     */
+    public void stop() {
+        if (running.compareAndSet(true, false)) {
+            log.info("Stopping Hiero observer: {}", getClass().getSimpleName());
+            executorService.shutdown();
+        }
+    }
+
+    private void pollSafe() {
+        try {
+            poll();
+        } catch (Exception e) {
+            log.error("Unexpected error during Hiero polling in {}", getClass().getSimpleName(), e);
+        }
+    }
+
+    /**
+     * The actual polling logic to be implemented by subclasses.
+     *
+     * @throws Exception if polling fails.
+     */
+    protected abstract void poll() throws Exception;
+
+    /**
+     * Notifies the listener of a new event.
+     *
+     * @param event The event to notify.
+     */
+    protected void notifyListener(@NonNull T event) {
+        try {
+            listener.onEvent(event);
+        } catch (Exception e) {
+            log.error("Error in event listener callback", e);
+        }
+    }
+}

hiero-enterprise-base/src/main/java/org/hiero/base/observer/EventObserver.java

@@ -0,0 +1,20 @@
+package org.hiero.base.observer;
+
+/**
+ * Functional interface for a component that observes events on the Hiero network.
+ *
+ * <p>Implementations of this interface typically run in a background thread or are
+ * triggered by a scheduler to check for new network activities.
+ *
+ * @param <T> The type of event being observed (e.g., TransactionInfo, TopicMessage).
+ */
+@FunctionalInterface
+public interface EventObserver<T> {
+
+    /**
+     * Called when a new event is detected.
+     *
+     * @param event The detected event.
+     */
+    void onEvent(T event);
+}

hiero-enterprise-base/src/main/java/org/hiero/base/observer/TopicObserver.java

@@ -0,0 +1,77 @@
+package org.hiero.base.observer;
+
+import com.hedera.hashgraph.sdk.TopicId;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Comparator;
+import java.util.List;
+import org.hiero.base.data.Page;
+import org.hiero.base.data.TopicMessage;
+import org.hiero.base.mirrornode.TopicRepository;
+import org.jspecify.annotations.NonNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Observer that polls for new messages on a specific Hiero Consensus Service (HCS) topic.
+ */
+public class TopicObserver extends AbstractPollingObserver<TopicMessage> {
+
+    private static final Logger log = LoggerFactory.getLogger(TopicObserver.class);
+
+    private final TopicRepository repository;
+    private final TopicId topicId;
+    private Instant lastSeenTimestamp;
+
+    /**
+     * Creates a new topic observer.
+     *
+     * @param repository The repository to use for polling.
+     * @param topicId The topic to monitor.
+     * @param pollingInterval How often to poll the mirror node.
+     * @param listener The callback for new messages.
+     */
+    public TopicObserver(
+            @NonNull TopicRepository repository,
+            @NonNull TopicId topicId,
+            @NonNull Duration pollingInterval,
+            @NonNull EventObserver<TopicMessage> listener) {
+        super(pollingInterval, listener);
+        this.repository = repository;
+        this.topicId = topicId;
+        this.lastSeenTimestamp = Instant.now();
+    }
+
+    @Override
+    protected void poll() throws Exception {
+        log.trace("Polling messages for topic {} after {}", topicId, lastSeenTimestamp);
+
+        Page<TopicMessage> page = repository.getMessages(topicId, lastSeenTimestamp);
+        List<TopicMessage> messages = page.getData();
+
+        if (messages.isEmpty()) {
+            return;
+        }
+
+        // Sort by timestamp to ensure chronological delivery
+        messages.sort(Comparator.comparing(TopicMessage::consensusTimestamp));
+
+        for (TopicMessage msg : messages) {
+            notifyListener(msg);
+            if (msg.consensusTimestamp().isAfter(lastSeenTimestamp)) {
+                lastSeenTimestamp = msg.consensusTimestamp();
+            }
+        }
+    }
+
+    /**
+     * Sets the starting timestamp for the observer.
+     *
+     * @param timestamp The timestamp to start from.
+     * @return this observer for chaining.
+     */
+    public TopicObserver startFrom(@NonNull Instant timestamp) {
+        this.lastSeenTimestamp = timestamp;
+        return this;
+    }
+}