Documentation

Backend pipeline with Spring Boot

Receive AI webhooks in Spring Boot controllers, process asynchronously with @Async, and call back to ModelRiver with WebClient.

Overview

Spring Boot's enterprise-grade features: dependency injection, async processing, and WebClient: provide a robust foundation for handling ModelRiver backend pipeline webhooks at scale.

What you'll build:

  • A REST controller endpoint for receiving webhooks
  • HMAC signature verification filter
  • Async service for background processing
  • Reactive WebClient for non-blocking callbacks

Quick start

Create project

Use Spring Initializr with:

  • Dependencies: Spring Web, Spring WebFlux (for WebClient)
  • Java version: 17+

Configuration

YAML
1# application.yml
2modelriver:
3 api-key: ${MODELRIVER_API_KEY}
4 webhook-secret: ${MODELRIVER_WEBHOOK_SECRET}
5 base-url: https://api.modelriver.com
JAVA
1// ModelRiverProperties.java
2@ConfigurationProperties(prefix = "modelriver")
3public record ModelRiverProperties(
4 String apiKey,
5 String webhookSecret,
6 String baseUrl
7) {}

Signature verification filter

JAVA
1// WebhookSignatureFilter.java
2import jakarta.servlet.*;
3import jakarta.servlet.http.*;
4import org.springframework.stereotype.Component;
5import org.springframework.web.util.ContentCachingRequestWrapper;
6import javax.crypto.Mac;
7import javax.crypto.spec.SecretKeySpec;
8import java.io.IOException;
9import java.nio.charset.StandardCharsets;
10import java.security.MessageDigest;
11 
12@Component
13public class WebhookSignatureFilter implements Filter {
14 
15 private final ModelRiverProperties properties;
16 
17 public WebhookSignatureFilter(ModelRiverProperties properties) {
18 this.properties = properties;
19 }
20 
21 @Override
22 public void doFilter(ServletRequest req, ServletResponse res, FilterChain chain)
23 throws IOException, ServletException {
24 HttpServletRequest request = (HttpServletRequest) req;
25 HttpServletResponse response = (HttpServletResponse) res;
26 
27 // Only apply to webhook endpoint
28 if (!request.getRequestURI().startsWith("/webhooks/modelriver")) {
29 chain.doFilter(req, res);
30 return;
31 }
32 
33 ContentCachingRequestWrapper wrappedRequest = new ContentCachingRequestWrapper(request);
34 chain.doFilter(wrappedRequest, res);
35 }
36 
37 public boolean verifySignature(String rawBody, String timestamp, String signature) {
38 // Signed string: "<timestamp>.<exact JSON text of the raw body's "data" value>" (never re-encode)
39 int start = rawBody.indexOf("\"data\":");
40 int end = rawBody.lastIndexOf(",\"timestamp\":");
41 if (start == -1 || end < start + 7) {
42 return false;
43 }
44 String data = rawBody.substring(start + 7, end);
45 
46 try {
47 Mac mac = Mac.getInstance("HmacSHA256");
48 SecretKeySpec key = new SecretKeySpec(
49 properties.webhookSecret().getBytes(StandardCharsets.UTF_8), "HmacSHA256"
50 );
51 mac.init(key);
52 byte[] hash = mac.doFinal((timestamp + "." + data).getBytes(StandardCharsets.UTF_8));
53 
54 String expected = bytesToHex(hash);
55 return MessageDigest.isEqual(
56 expected.getBytes(StandardCharsets.UTF_8),
57 signature.getBytes(StandardCharsets.UTF_8)
58 );
59 } catch (Exception e) {
60 return false;
61 }
62 }
63 
64 private String bytesToHex(byte[] bytes) {
65 StringBuilder sb = new StringBuilder();
66 for (byte b : bytes) {
67 sb.append(String.format("%02x", b));
68 }
69 return sb.toString();
70 }
71}

Webhook controller

JAVA
1// WebhookController.java
2import org.springframework.http.ResponseEntity;
3import org.springframework.web.bind.annotation.*;
4import java.nio.charset.StandardCharsets;
5import java.util.Map;
6 
7@RestController
8@RequestMapping("/webhooks")
9public class WebhookController {
10 
11 private final WebhookSignatureFilter signatureFilter;
12 private final AiWebhookProcessor processor;
13 
14 public WebhookController(
15 WebhookSignatureFilter signatureFilter,
16 AiWebhookProcessor processor
17 ) {
18 this.signatureFilter = signatureFilter;
19 this.processor = processor;
20 }
21 
22 @PostMapping("/modelriver")
23 @SuppressWarnings("unchecked")
24 public ResponseEntity<Map<String, Boolean>> handleWebhook(
25 @RequestBody byte[] body,
26 @RequestHeader(value = "x-modelriver-signature", defaultValue = "") String signature,
27 @RequestHeader(value = "x-modelriver-timestamp", defaultValue = "") String timestamp
28 ) {
29 String rawBody = new String(body, StandardCharsets.UTF_8); // exact raw body
30 
31 // 1. Verify signature
32 if (!signatureFilter.verifySignature(rawBody, timestamp, signature)) {
33 return ResponseEntity.status(401).build();
34 }
35 
36 // 2. Parse payload: the event payload is under "data"
37 Map<String, Object> payload =
38 (Map<String, Object>) parseJson(rawBody).getOrDefault("data", Map.of());
39 String type = (String) payload.getOrDefault("type", "");
40 String callbackUrl = (String) payload.get("callback_url");
41 
42 // 3. Handle backend pipeline workflow
43 if ("task.ai_generated".equals(type) && callbackUrl != null) {
44 processor.processAsync(
45 (String) payload.get("event"),
46 (Map<String, Object>) payload.get("ai_response"),
47 callbackUrl,
48 (Map<String, Object>) payload.getOrDefault("customer_data", Map.of())
49 );
50 return ResponseEntity.ok(Map.of("received", true));
51 }
52 
53 return ResponseEntity.ok(Map.of("received", true));
54 }
55 
56 @SuppressWarnings("unchecked")
57 private Map<String, Object> parseJson(String json) {
58 try {
59 return new com.fasterxml.jackson.databind.ObjectMapper()
60 .readValue(json, Map.class);
61 } catch (Exception e) {
62 return Map.of();
63 }
64 }
65}

Async processor service

JAVA
1// AiWebhookProcessor.java
2import org.springframework.scheduling.annotation.Async;
3import org.springframework.stereotype.Service;
4import org.springframework.web.reactive.function.client.WebClient;
5import org.slf4j.Logger;
6import org.slf4j.LoggerFactory;
7import java.time.Instant;
8import java.util.HashMap;
9import java.util.Map;
10 
11@Service
12public class AiWebhookProcessor {
13 
14 private static final Logger log = LoggerFactory.getLogger(AiWebhookProcessor.class);
15 private final WebClient webClient;
16 private final ModelRiverProperties properties;
17 
18 public AiWebhookProcessor(ModelRiverProperties properties) {
19 this.properties = properties;
20 this.webClient = WebClient.builder()
21 .defaultHeader("Content-Type", "application/json")
22 .build();
23 }
24 
25 @Async
26 public void processAsync(
27 String event,
28 Map<String, Object> aiResponse,
29 String callbackUrl,
30 Map<String, Object> customerData
31 ) {
32 try {
33 Map<String, Object> data = aiResponse != null
34 ? new HashMap<>((Map<String, Object>) aiResponse.getOrDefault("data", Map.of()))
35 : new HashMap<>();
36 
37 // Your custom business logic
38 if ("content_ready".equals(event)) {
39 data.put("processed", true);
40 data.put("saved_at", Instant.now().toString());
41 }
42 
43 // Call back to ModelRiver
44 Map<String, Object> callback = Map.of(
45 "data", data,
46 "task_id", "springboot_" + event + "_" + Instant.now().getEpochSecond(),
47 "metadata", Map.of(
48 "processed_by", "spring-boot",
49 "processed_at", Instant.now().toString()
50 )
51 );
52 
53 webClient.post()
54 .uri(callbackUrl)
55 .header("Authorization", "Bearer " + properties.apiKey())
56 .bodyValue(callback)
57 .retrieve()
58 .toBodilessEntity()
59 .block();
60 
61 log.info("✅ Callback sent for event: {}", event);
62 
63 } catch (Exception e) {
64 log.error("❌ Callback failed: {}", e.getMessage());
65 
66 // Send error callback
67 try {
68 webClient.post()
69 .uri(callbackUrl)
70 .header("Authorization", "Bearer " + properties.apiKey())
71 .bodyValue(Map.of(
72 "data", Map.of(
73 "processing_status", "failed",
74 "processing_error", String.valueOf(e.getMessage())
75 )
76 ))
77 .retrieve()
78 .toBodilessEntity()
79 .block();
80 } catch (Exception ignored) {}
81 }
82 }
83}

Enable async processing:

JAVA
1// Application.java
2@SpringBootApplication
3@EnableAsync
4@EnableConfigurationProperties(ModelRiverProperties.class)
5public class MyAiAppApplication {
6 public static void main(String[] args) {
7 SpringApplication.run(MyAiAppApplication.class, args);
8 }
9}

Best practices

  1. Use @Async: Process webhooks in a separate thread pool to avoid blocking.
  2. Use WebClient: Non-blocking HTTP client for callbacks.
  3. Use @ConfigurationProperties: Type-safe configuration management.
  4. Use MessageDigest.isEqual: Constant-time comparison for signature verification.
  5. Configure thread pool: Set spring.task.execution.pool.core-size for production loads.

Next steps