[#22] On-Chain Event Ingestion Pipeline with Transactional Outbox & Replay

backend-api/src/main/java/com/guildworkman/api/GuildWorkmanApplication.java

@@ -2,10 +2,12 @@
 
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.scheduling.annotation.EnableScheduling;
 
 import java.time.LocalDateTime;
 
 @SpringBootApplication
+@EnableScheduling
 public class GuildWorkmanApplication {
 
 	public static void main(String[] args) {

backend-api/src/main/java/com/guildworkman/api/chain/api/ChainEventController.java

@@ -0,0 +1,21 @@
+package com.guildworkman.api.chain.api;
+
+import com.guildworkman.api.chain.service.ChainEventService;
+import jakarta.validation.Valid;
+import lombok.RequiredArgsConstructor;
+import org.springframework.http.HttpStatus;
+import org.springframework.web.bind.annotation.*;
+
+import java.util.Map;
+
+@RestController @RequestMapping("/api/v1/chain/events") @RequiredArgsConstructor
+public class ChainEventController {
+    private final ChainEventService service;
+
+    @PostMapping
+    @ResponseStatus(HttpStatus.ACCEPTED)
+    public ChainEventResponse ingest(@Valid @RequestBody IngestChainEventRequest request) { return service.ingest(request); }
+
+    @PostMapping("/replay")
+    public Map<String, Integer> replay(@Valid @RequestBody ReplayRequest request) { return Map.of("replayed", service.replay(request)); }
+}

backend-api/src/main/java/com/guildworkman/api/chain/api/ChainEventResponse.java

@@ -0,0 +1,8 @@
+package com.guildworkman.api.chain.api;
+
+import com.guildworkman.api.chain.model.OnChainEvent;
+import java.time.Instant;
+
+public record ChainEventResponse(Long id, String eventKey, String contractId, long ledger, int eventIndex, String topics, String payload, String status, int attempts, Instant processedAt) {
+    public static ChainEventResponse from(OnChainEvent e) { return new ChainEventResponse(e.getId(), e.getEventKey(), e.getContractId(), e.getLedger(), e.getEventIndex(), e.getTopics(), e.getPayload(), e.getStatus().name(), e.getAttempts(), e.getProcessedAt()); }
+}

backend-api/src/main/java/com/guildworkman/api/chain/api/IngestChainEventRequest.java

@@ -0,0 +1,12 @@
+package com.guildworkman.api.chain.api;
+
+import jakarta.validation.constraints.*;
+import java.util.List;
+
+public record IngestChainEventRequest(
+        @NotBlank @Size(max = 128) String eventKey,
+        @NotBlank @Size(max = 128) String contractId,
+        @PositiveOrZero long ledger,
+        @PositiveOrZero int eventIndex,
+        @NotEmpty List<@NotBlank @Size(max = 128) String> topics,
+        @NotBlank String payload) {}

backend-api/src/main/java/com/guildworkman/api/chain/api/ReplayRequest.java

@@ -0,0 +1,7 @@
+package com.guildworkman.api.chain.api;
+
+import jakarta.validation.constraints.PositiveOrZero;
+
+public record ReplayRequest(@PositiveOrZero long fromLedger, @PositiveOrZero long toLedger) {
+    public ReplayRequest { if (toLedger < fromLedger) throw new IllegalArgumentException("toLedger must be greater than or equal to fromLedger"); }
+}

backend-api/src/main/java/com/guildworkman/api/chain/model/ChainEventStatus.java

@@ -0,0 +1,5 @@
+package com.guildworkman.api.chain.model;
+
+public enum ChainEventStatus {
+    PENDING, PROCESSING, PROCESSED, DEAD_LETTER
+}

backend-api/src/main/java/com/guildworkman/api/chain/model/OnChainEvent.java

@@ -0,0 +1,46 @@
+package com.guildworkman.api.chain.model;
+
+import jakarta.persistence.*;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+import java.time.Instant;
+
+@Entity
+@Table(name = "on_chain_events", uniqueConstraints = @UniqueConstraint(name = "uk_chain_event_key", columnNames = "event_key"), indexes = {
+        @Index(name = "idx_chain_event_stream_order", columnList = "contract_id,ledger,event_index"),
+        @Index(name = "idx_chain_event_status", columnList = "status,next_attempt_at")
+})
+@Getter
+@Setter
+@NoArgsConstructor
+public class OnChainEvent {
+    @Id @GeneratedValue(strategy = GenerationType.IDENTITY)
+    private Long id;
+    @Column(name = "event_key", nullable = false, updatable = false, length = 128)
+    private String eventKey;
+    @Column(name = "contract_id", nullable = false, length = 128)
+    private String contractId;
+    @Column(nullable = false)
+    private long ledger;
+    @Column(name = "event_index", nullable = false)
+    private int eventIndex;
+    @Column(nullable = false, length = 256)
+    private String topics;
+    @Lob @Column(nullable = false)
+    private String payload;
+    @Enumerated(EnumType.STRING) @Column(nullable = false, length = 20)
+    private ChainEventStatus status = ChainEventStatus.PENDING;
+    @Column(nullable = false)
+    private int attempts;
+    @Column(name = "next_attempt_at", nullable = false)
+    private Instant nextAttemptAt = Instant.now();
+    @Column(length = 1000)
+    private String lastError;
+    @Column(nullable = false, updatable = false)
+    private Instant createdAt = Instant.now();
+    private Instant processedAt;
+    @Version
+    private long version;
+}

backend-api/src/main/java/com/guildworkman/api/chain/model/OutboxEvent.java

@@ -0,0 +1,31 @@
+package com.guildworkman.api.chain.model;
+
+import jakarta.persistence.*;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+import java.time.Instant;
+
+@Entity
+@Table(name = "chain_event_outbox", uniqueConstraints = @UniqueConstraint(name = "uk_outbox_event", columnNames = "event_id"), indexes = @Index(name = "idx_outbox_status", columnList = "status,next_attempt_at"))
+@Getter @Setter @NoArgsConstructor
+public class OutboxEvent {
+    @Id @GeneratedValue(strategy = GenerationType.IDENTITY)
+    private Long id;
+    @Column(name = "event_id", nullable = false, updatable = false)
+    private Long eventId;
+    @Enumerated(EnumType.STRING) @Column(nullable = false, length = 20)
+    private OutboxStatus status = OutboxStatus.PENDING;
+    @Column(nullable = false)
+    private int attempts;
+    @Column(name = "next_attempt_at", nullable = false)
+    private Instant nextAttemptAt = Instant.now();
+    @Column(length = 1000)
+    private String lastError;
+    @Column(nullable = false, updatable = false)
+    private Instant createdAt = Instant.now();
+    private Instant completedAt;
+    @Version
+    private long version;
+}

backend-api/src/main/java/com/guildworkman/api/chain/model/OutboxStatus.java

@@ -0,0 +1,5 @@
+package com.guildworkman.api.chain.model;
+
+public enum OutboxStatus {
+    PENDING, PROCESSING, COMPLETED, DEAD_LETTER
+}

backend-api/src/main/java/com/guildworkman/api/chain/repository/OnChainEventRepository.java

@@ -0,0 +1,21 @@
+package com.guildworkman.api.chain.repository;
+
+import com.guildworkman.api.chain.model.ChainEventStatus;
+import com.guildworkman.api.chain.model.OnChainEvent;
+import org.springframework.data.domain.Pageable;
+import org.springframework.data.jpa.repository.*;
+import org.springframework.data.repository.query.Param;
+
+import jakarta.persistence.LockModeType;
+import java.time.Instant;
+import java.util.*;
+
+public interface OnChainEventRepository extends JpaRepository<OnChainEvent, Long> {
+    Optional<OnChainEvent> findByEventKey(String eventKey);
+    boolean existsByEventKey(String eventKey);
+    @Lock(LockModeType.PESSIMISTIC_WRITE)
+    @Query("select e from OnChainEvent e where e.status in :statuses and e.nextAttemptAt <= :now order by e.contractId, e.ledger, e.eventIndex, e.id")
+    List<OnChainEvent> claimNext(@Param("statuses") Set<ChainEventStatus> statuses, @Param("now") Instant now, Pageable pageable);
+    List<OnChainEvent> findByLedgerBetweenOrderByContractIdAscLedgerAscEventIndexAsc(long fromLedger, long toLedger);
+    long countByStatus(ChainEventStatus status);
+}

backend-api/src/main/java/com/guildworkman/api/chain/repository/OutboxEventRepository.java

@@ -0,0 +1,16 @@
+package com.guildworkman.api.chain.repository;
+
+import com.guildworkman.api.chain.model.*;
+import org.springframework.data.domain.Pageable;
+import org.springframework.data.jpa.repository.*;
+import org.springframework.data.repository.query.Param;
+import jakarta.persistence.LockModeType;
+import java.time.Instant;
+import java.util.*;
+
+public interface OutboxEventRepository extends JpaRepository<OutboxEvent, Long> {
+    @Lock(LockModeType.PESSIMISTIC_WRITE)
+    @Query("select o from OutboxEvent o where o.status in :statuses and o.nextAttemptAt <= :now order by o.id")
+    List<OutboxEvent> claimNext(@Param("statuses") Set<OutboxStatus> statuses, @Param("now") Instant now, Pageable pageable);
+    Optional<OutboxEvent> findByEventId(Long eventId);
+}

backend-api/src/main/java/com/guildworkman/api/chain/service/ChainEventHandler.java

@@ -0,0 +1,8 @@
+package com.guildworkman.api.chain.service;
+
+import com.guildworkman.api.chain.model.OnChainEvent;
+
+@FunctionalInterface
+public interface ChainEventHandler {
+    void handle(OnChainEvent event);
+}

backend-api/src/main/java/com/guildworkman/api/chain/service/ChainEventInserter.java

@@ -0,0 +1,49 @@
+package com.guildworkman.api.chain.service;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.guildworkman.api.chain.api.IngestChainEventRequest;
+import com.guildworkman.api.chain.model.ChainEventStatus;
+import com.guildworkman.api.chain.model.OnChainEvent;
+import com.guildworkman.api.chain.model.OutboxEvent;
+import com.guildworkman.api.chain.repository.OnChainEventRepository;
+import com.guildworkman.api.chain.repository.OutboxEventRepository;
+import lombok.RequiredArgsConstructor;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Propagation;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.time.Instant;
+
+/**
+ * Isolated insert so a unique-key race aborts only this nested transaction
+ * (Postgres), leaving the caller's transaction able to re-read the winner.
+ */
+@Service
+@RequiredArgsConstructor
+public class ChainEventInserter {
+    private final OnChainEventRepository events;
+    private final OutboxEventRepository outbox;
+    private final ObjectMapper objectMapper;
+
+    @Transactional(propagation = Propagation.REQUIRES_NEW)
+    public OnChainEvent insert(IngestChainEventRequest request) {
+        OnChainEvent event = new OnChainEvent();
+        event.setEventKey(request.eventKey());
+        event.setContractId(request.contractId());
+        event.setLedger(request.ledger());
+        event.setEventIndex(request.eventIndex());
+        try {
+            event.setTopics(objectMapper.writeValueAsString(request.topics()));
+        } catch (Exception ex) {
+            throw new IllegalArgumentException("topics must be serializable", ex);
+        }
+        event.setPayload(request.payload());
+        event.setStatus(ChainEventStatus.PENDING);
+        event.setNextAttemptAt(Instant.now());
+        OnChainEvent saved = events.saveAndFlush(event);
+        OutboxEvent message = new OutboxEvent();
+        message.setEventId(saved.getId());
+        outbox.saveAndFlush(message);
+        return saved;
+    }
+}

backend-api/src/main/java/com/guildworkman/api/chain/service/ChainEventService.java

@@ -0,0 +1,111 @@
+package com.guildworkman.api.chain.service;
+
+import com.guildworkman.api.chain.api.*;
+import com.guildworkman.api.chain.model.*;
+import com.guildworkman.api.chain.repository.*;
+import lombok.RequiredArgsConstructor;
+import org.springframework.dao.DataIntegrityViolationException;
+import org.springframework.data.domain.PageRequest;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.time.Instant;
+import java.util.EnumSet;
+import java.util.List;
+
+@Service
+@RequiredArgsConstructor
+public class ChainEventService {
+    private final OnChainEventRepository events;
+    private final OutboxEventRepository outbox;
+    private final List<ChainEventHandler> handlers;
+    private final ChainEventInserter inserter;
+
+    static final int MAX_ATTEMPTS = 5;
+
+    @Transactional(readOnly = true)
+    public ChainEventResponse ingest(IngestChainEventRequest request) {
+        return events.findByEventKey(request.eventKey())
+                .map(ChainEventResponse::from)
+                .orElseGet(() -> insertIdempotently(request));
+    }
+
+    private ChainEventResponse insertIdempotently(IngestChainEventRequest request) {
+        try {
+            return ChainEventResponse.from(inserter.insert(request));
+        } catch (DataIntegrityViolationException ex) {
+            // Nested REQUIRES_NEW insert rolled back; outer TX can still read the winner.
+            return events.findByEventKey(request.eventKey())
+                    .map(ChainEventResponse::from)
+                    .orElseThrow(() -> new IllegalStateException(
+                            "Event not found after idempotent-guard violation for key=" + request.eventKey(), ex));
+        }
+    }
+
+    @Transactional
+    public int replay(ReplayRequest request) {
+        int count = 0;
+        List<OnChainEvent> batch = events.findByLedgerBetweenOrderByContractIdAscLedgerAscEventIndexAsc(
+                request.fromLedger(), request.toLedger());
+        Instant now = Instant.now();
+        for (OnChainEvent event : batch) {
+            event.setStatus(ChainEventStatus.PENDING);
+            event.setAttempts(0);
+            event.setLastError(null);
+            event.setProcessedAt(null);
+            event.setNextAttemptAt(now);
+            events.save(event);
+
+            outbox.findByEventId(event.getId()).ifPresent(message -> {
+                message.setStatus(OutboxStatus.PENDING);
+                message.setAttempts(0);
+                message.setLastError(null);
+                message.setCompletedAt(null);
+                message.setNextAttemptAt(now);
+                outbox.save(message);
+            });
+            count++;
+        }
+        return count;
+    }
+
+    @Scheduled(fixedDelayString = "${chain.events.poll-delay-ms:1000}")
+    @Transactional
+    public void processOne() {
+        events.claimNext(
+                EnumSet.of(ChainEventStatus.PENDING, ChainEventStatus.PROCESSING),
+                Instant.now(),
+                PageRequest.of(0, 1)
+        ).stream().findFirst().ifPresent(this::process);
+    }
+
+    void process(OnChainEvent event) {
+        try {
+            event.setStatus(ChainEventStatus.PROCESSING);
+            event.setAttempts(event.getAttempts() + 1);
+            for (ChainEventHandler handler : handlers) {
+                handler.handle(event);
+            }
+            event.setStatus(ChainEventStatus.PROCESSED);
+            event.setProcessedAt(Instant.now());
+            event.setLastError(null);
+            events.save(event);
+            outbox.findByEventId(event.getId()).ifPresent(message -> {
+                message.setStatus(OutboxStatus.COMPLETED);
+                message.setCompletedAt(Instant.now());
+                message.setLastError(null);
+                outbox.save(message);
+            });
+        } catch (RuntimeException ex) {
+            event.setLastError(ex.getMessage());
+            if (event.getAttempts() >= MAX_ATTEMPTS) {
+                event.setStatus(ChainEventStatus.DEAD_LETTER);
+            } else {
+                event.setStatus(ChainEventStatus.PENDING);
+                event.setNextAttemptAt(Instant.now().plusSeconds(1L << Math.min(event.getAttempts(), 6)));
+            }
+            events.save(event);
+        }
+    }
+}

backend-api/src/main/java/com/guildworkman/api/config/SecurityConfig.java

@@ -58,6 +58,7 @@ public class SecurityConfig {
             "/api/v1/auth/logout",
             "/api/v1/client/**",
             "/api/v1/skilledWorker/**",
+            "/api/v1/chain/events/**",
             "/v3/api-docs/**",
             "/swagger-ui/**",
             "/swagger-ui.html"

backend-api/src/main/resources/application.properties

@@ -33,6 +33,9 @@ spring.mvc.problemdetails.enabled=true
 # staging/prod can point at their own canonical docs host.
 guildworkman.problem-details.type-base=${PROBLEM_TYPE_BASE:https://guildworkman.dev/problems/}
 
+# On-chain ingestion worker. Set to 0 to disable polling in a batch/replay job.
+chain.events.poll-delay-ms=${CHAIN_EVENTS_POLL_DELAY_MS:1000}
+
 # JWT authentication.
 #   - secret: HMAC-256 signing key. MUST be overridden in every real
 #     environment via JWT_SECRET (>= 32 bytes). The baked-in default exists