Augment catchup latest (#48)

* [augment] Augment adding fetch latest

* [augment] Augment forgot server implementation

* [human] Off by 1 error

* [human] Switch writing back on, change db type

Author: p14n

app/infra/tf/rds.tf

@@ -30,7 +30,8 @@ resource "aws_db_instance" "postevent" {
   identifier                   = "postevent-${count.index}"
   engine                       = "postgres"
   engine_version               = "17"
-  instance_class               = "db.t3.micro"
+  instance_class               = "db.t4g.micro"
+  storage_type                 = "gp3"
   allocated_storage            = 20
   storage_encrypted            = true
   performance_insights_enabled = true

app/src/main/java/com/p14n/postevent/App.java

@@ -139,8 +139,7 @@ private static void run(String affinity, String[] write, String[] read, String d
                 }
 
             } else {
-                Thread.currentThread().join();
-                // writeContinuously(ds, affinity, write, ot);
+                writeContinuously(ds, affinity, write, ot);
             }
 
         } finally {
@@ -168,8 +167,9 @@ private static NettyChannelBuilder buildClientChannel(String host, int port) {
                 .sslContext(buildSslContext());
     }
 
-    private static RemotePersistentConsumer runConsumerClient(String[] write, String[] read, String[] topichosts, DataSource ds,
-                                                              OpenTelemetry ot) {
+    private static RemotePersistentConsumer runConsumerClient(String[] write, String[] read, String[] topichosts,
+            DataSource ds,
+            OpenTelemetry ot) {
 
         RemotePersistentConsumer cc;
         cc = new RemotePersistentConsumer(ot, 10);

library/src/main/java/com/p14n/postevent/LocalPersistentConsumer.java

@@ -58,12 +58,15 @@ public void start() throws IOException, InterruptedException {
             var lc = new LocalConsumer<>(cfg, pb);
 
             seb.subscribe(new CatchupService(ds, new CatchupServer(ds), seb));
-            var unprocessedSubmitter = new UnprocessedSubmitter(seb,ds, new UnprocessedEventFinder(), tb, batchSize);
+            var unprocessedSubmitter = new UnprocessedSubmitter(seb, ds, new UnprocessedEventFinder(), tb, batchSize);
             seb.subscribe(unprocessedSubmitter);
 
             asyncExecutor.scheduleAtFixedRate(() -> {
-                logger.atDebug().log("Triggering unprocessed check");
+                logger.atDebug().log("Triggering unprocessed check and fetch latest");
                 seb.publish(SystemEvent.UnprocessedCheckRequired);
+                for (String topic : cfg.topics()) {
+                    seb.publish(SystemEvent.FetchLatest.withTopic(topic));
+                }
             }, 30, 30, TimeUnit.SECONDS);
 
             lc.start();

library/src/main/java/com/p14n/postevent/RemotePersistentConsumer.java

@@ -76,10 +76,15 @@ public void start(Set<String> topics, DataSource ds, ManagedChannel channel) {
                 client.subscribe(topic, pb);
             }
             seb.subscribe(new CatchupService(ds, catchupClient, seb));
-            seb.subscribe(new UnprocessedSubmitter(seb,ds, new UnprocessedEventFinder(), tb, batchSize));
+            seb.subscribe(new UnprocessedSubmitter(seb, ds, new UnprocessedEventFinder(), tb, batchSize));
 
             asyncExecutor.scheduleAtFixedRate(
-                    () -> seb.publish(SystemEvent.UnprocessedCheckRequired),
+                    () -> {
+                        seb.publish(SystemEvent.UnprocessedCheckRequired);
+                        for (String topic : topics) {
+                            seb.publish(SystemEvent.FetchLatest.withTopic(topic));
+                        }
+                    },
                     30, 30, TimeUnit.SECONDS);
 
             closeables = List.of(client, catchupClient, pb, seb, tb);

library/src/main/java/com/p14n/postevent/broker/SystemEvent.java

@@ -5,7 +5,8 @@
 public enum SystemEvent implements Traceable {
 
     CatchupRequired,
-    UnprocessedCheckRequired;
+    UnprocessedCheckRequired,
+    FetchLatest; // New event type
 
     public String topic;
 

library/src/main/java/com/p14n/postevent/catchup/CatchupServer.java

@@ -65,4 +65,27 @@ public List<Event> fetchEvents(long startAfter, long end, int maxResults, String
             throw new RuntimeException("Failed to fetch events", e);
         }
     }
-}
+
+    @Override
+    public long getLatestMessageId(String topic) {
+        if (topic == null || topic.trim().isEmpty()) {
+            throw new IllegalArgumentException("Topic name cannot be null or empty");
+        }
+
+        String sql = String.format("SELECT MAX(idn) FROM postevent.%s", topic);
+
+        try (Connection conn = dataSource.getConnection();
+                PreparedStatement stmt = conn.prepareStatement(sql);
+                ResultSet rs = stmt.executeQuery()) {
+
+            if (rs.next()) {
+                return rs.getLong(1);
+            }
+            return 0;
+
+        } catch (SQLException e) {
+            logger.atError().setCause(e).log("Error fetching latest message ID");
+            throw new RuntimeException("Failed to fetch latest message ID", e);
+        }
+    }
+}

library/src/main/java/com/p14n/postevent/catchup/CatchupServerInterface.java

@@ -17,4 +17,12 @@ public interface CatchupServerInterface {
      * @return A list of events within the specified range
      */
     List<Event> fetchEvents(long startAfter, long end, int maxResults, String topic);
-}
+
+    /**
+     * Retrieves the latest message ID for a given topic.
+     *
+     * @param topic The topic to get the latest message ID for
+     * @return The latest message ID for the specified topic
+     */
+    long getLatestMessageId(String topic);
+}

library/src/main/java/com/p14n/postevent/catchup/CatchupService.java

@@ -186,6 +186,66 @@ private void updateHwm(Connection connection, String topicName, long currentHwm,
     public void onMessage(SystemEvent message) {
         if (Objects.requireNonNull(message) == SystemEvent.CatchupRequired) {
             oneAtATime(() -> catchup(message.topic), () -> onMessage(message));
+        } else if (message == SystemEvent.FetchLatest) {
+            oneAtATime(() -> fetchLatest(message.topic), () -> onMessage(message));
+        }
+    }
+
+    private int fetchLatest(String topicName) {
+        if (topicName == null) {
+            LOGGER.warn("Topic name is null for fetch latest");
+            return 0;
+        }
+
+        try (Connection conn = datasource.getConnection()) {
+            conn.setAutoCommit(false);
+
+            long currentHwm = getCurrentHwm(conn, topicName);
+
+            // Get the latest message ID from the server
+            long latestMessageId = catchupServer.getLatestMessageId(topicName);
+
+            if (latestMessageId <= currentHwm) {
+                LOGGER.info("No new messages found after HWM {} for topic {}", currentHwm, topicName);
+                return 0;
+            }
+
+            // Fetch just the latest message
+            List<Event> events = catchupServer.fetchEvents(latestMessageId - 1, latestMessageId, 1, topicName);
+
+            if (events.isEmpty()) {
+                LOGGER.info("No events found in range for topic: " + topicName);
+                return 0;
+            }
+
+            // Write event to messages table
+            int processedCount = writeEventsToMessagesTable(conn, events);
+
+            LOGGER.info("Processed latest event for topic {}",
+                    topicName);
+
+            conn.commit();
+
+            // If there are more messages between currentHwm and latestMessageId,
+            // trigger a catchup to get the rest
+            systemEventBroker.publish(SystemEvent.CatchupRequired.withTopic(topicName));
+
+            return processedCount;
+        } catch (SQLException e) {
+            LOGGER.error("Failed to fetch latest", e);
+            return 0;
+        }
+    }
+
+    private long findLatestMessageId(Connection connection, String topicName) throws SQLException {
+        String sql = "SELECT MAX(idn) FROM postevent." + topicName;
+        try (PreparedStatement stmt = connection.prepareStatement(sql)) {
+            try (ResultSet rs = stmt.executeQuery()) {
+                if (rs.next()) {
+                    return rs.getLong(1);
+                }
+                return 0;
+            }
         }
     }
 
@@ -291,4 +351,4 @@ public AtomicBoolean getRunning() {
         return running;
     }
 
-}
+}

library/src/main/java/com/p14n/postevent/catchup/grpc/CatchupGrpcClient.java

@@ -85,6 +85,26 @@ private Event convertFromGrpcEvent(com.p14n.postevent.catchup.grpc.Event grpcEve
                 grpcEvent.getTraceparent());
     }
 
+    @Override
+    public long getLatestMessageId(String topic) {
+        logger.atInfo()
+                .addArgument(topic)
+                .log("Fetching latest message ID for topic {}");
+
+        TopicRequest request = TopicRequest.newBuilder()
+                .setTopic(topic)
+                .build();
+
+        LatestMessageIdResponse response;
+        try {
+            response = blockingStub.getLatestMessageId(request);
+            return response.getMessageId();
+        } catch (StatusRuntimeException e) {
+            logger.atWarn().setCause(e).log("RPC failed: {}", e.getStatus());
+            throw new RuntimeException("Failed to fetch latest message ID via gRPC", e);
+        }
+    }
+
     @Override
     public void close() {
         try {

library/src/main/java/com/p14n/postevent/catchup/grpc/CatchupGrpcServer.java

@@ -59,6 +59,23 @@ public CatchupServiceImpl(CatchupServerInterface catchupServer) {
             this.catchupServer = catchupServer;
         }
 
+        @Override
+        public void getLatestMessageId(TopicRequest request, StreamObserver<LatestMessageIdResponse> responseObserver) {
+            try {
+                long latestId = catchupServer.getLatestMessageId(request.getTopic());
+
+                LatestMessageIdResponse response = LatestMessageIdResponse.newBuilder()
+                        .setMessageId(latestId)
+                        .build();
+
+                responseObserver.onNext(response);
+                responseObserver.onCompleted();
+            } catch (Exception e) {
+                logger.error("Error getting latest message ID", e);
+                responseObserver.onError(e);
+            }
+        }
+
         @Override
         public void fetchEvents(FetchEventsRequest request, StreamObserver<FetchEventsResponse> responseObserver) {
             try {