Programmatic Access

// Token provider — caches the JWT and refreshes on expiry.
@Component
public class MusubiAuthProvider {
  private final WebClient auth;
  private final AtomicReference<Token> cached = new AtomicReference<>();

  public record Token(String value, Instant expiresAt) {}

  public MusubiAuthProvider(WebClient.Builder b, @Value("${musubi.backend-url}") String url) {
    this.auth = b.baseUrl(url).build();
  }

  public String token() {
    var t = cached.get();
    if (t != null && Instant.now().isBefore(t.expiresAt().minusSeconds(60))) return t.value();
    return refresh();
  }

  private synchronized String refresh() {
    var t = auth.post().uri("/auth/token")
        .bodyValue(Map.of(/* your IdP credentials */))
        .retrieve().bodyToMono(Token.class).block();
    cached.set(t);
    return t.value();
  }
}

// Typed DTOs. Match the shapes in /custodian/api-reference.
public record ApiResponse<T>(T data) {}
public record OrderDto(
    String intentId, String status, String senderPartyId,
    String sourceCurrency, String targetCurrency, String targetAmount,
    String sourceAmountMax, Instant createdAt, Instant expiresAt) {}
public record QuoteDto(
    String quoteId, String intentId, String marketMakerPartyId,
    String fxRate, String sourceAmount, String targetAmount,
    Instant submittedAt, Instant validUntil, String status) {}

// Service — list proposals, review quotes, accept.
@Service
public class MusubiCustodianClient {
  private final WebClient rest;

  public MusubiCustodianClient(WebClient.Builder b,
                               @Value("${musubi.backend-url}") String url,
                               MusubiAuthProvider auth) {
    this.rest = b.baseUrl(url)
        .filter((req, next) -> next.exchange(
            ClientRequest.from(req).headers(h -> h.setBearerAuth(auth.token())).build()))
        .build();
  }

  public List<OrderDto> listPending() {
    return rest.get().uri("/api/v1/orders?status=PENDING")
        .retrieve()
        .bodyToMono(new ParameterizedTypeReference<ApiResponse<List<OrderDto>>>() {})
        .map(ApiResponse::data)
        .block();
  }

  public List<QuoteDto> listQuotes(String intentId) {
    return rest.get().uri("/api/v1/orders/{id}/quotes", intentId)
        .retrieve()
        .bodyToMono(new ParameterizedTypeReference<ApiResponse<List<QuoteDto>>>() {})
        .map(ApiResponse::data)
        .block();
  }

  public OrderDto acceptProposal(String intentId) {
    return rest.post().uri("/api/v1/orders/{id}/accept", intentId)
        .retrieve()
        .bodyToMono(new ParameterizedTypeReference<ApiResponse<OrderDto>>() {})
        .map(ApiResponse::data)
        .block();
  }

  public OrderDto acceptQuote(String intentId, String quoteId) {
    // Co-signs a pending QuoteAcceptanceProposal — the institution emitted
    // it via its own POST .../accept (returns 202). This call exercises
    // AcceptQuoteProposal: atomic FXOrder.AcceptQuote (PENDING → QUOTED)
    // + FXQuote.Accept on the winner + QuoteRequest.DeactivateRequest +
    // FXQuote.RejectQuote on losers. Returns 200 + the QUOTED FXOrder.
    return rest.post().uri("/api/v1/orders/{id}/quotes/{qid}/accept", intentId, quoteId)
        .retrieve()
        .bodyToMono(new ParameterizedTypeReference<ApiResponse<OrderDto>>() {})
        .map(ApiResponse::data)
        .block();
  }

  public RejectAck rejectQuote(String intentId, String quoteId, String rejectionDetail) {
    // Declines the pending QuoteAcceptanceProposal — β quote-tainting:
    // the proposed FXQuote becomes terminal QUOTE_REJECTED. Other PENDING
    // quotes are unaffected; the institution can propose a different one.
    // Server enforces rejection_detail >= 20 chars (DAML ensure + service
    // guard) — backend returns 422 if shorter.
    return rest.post().uri("/api/v1/orders/{id}/quotes/{qid}/reject", intentId, quoteId)
        .body(Map.of("rejection_detail", rejectionDetail))
        .retrieve()
        .bodyToMono(new ParameterizedTypeReference<ApiResponse<RejectAck>>() {})
        .map(ApiResponse::data)
        .block();
  }
}

public record OrderEvent(
    String intentId, String status, String transactionHash,
    Instant settledAt, String sourceAmountActual, String targetAmount,
    String fxRate) {}

@Component
public class MusubiSettlementListener {
  private static final Logger log = LoggerFactory.getLogger(MusubiSettlementListener.class);

  private final WebClient rest;
  private final AccountingService accounting;

  public MusubiSettlementListener(WebClient.Builder b,
                                  @Value("${musubi.backend-url}") String url,
                                  MusubiAuthProvider auth,
                                  AccountingService accounting) {
    this.rest = b.baseUrl(url)
        .filter((req, next) -> next.exchange(
            ClientRequest.from(req).headers(h -> h.setBearerAuth(auth.token())).build()))
        .build();
    this.accounting = accounting;
  }

  @EventListener(ApplicationReadyEvent.class)
  public void subscribe() {
    var type = new ParameterizedTypeReference<ServerSentEvent<OrderEvent>>() {};
    rest.get().uri("/api/v1/orders/events")
        .accept(MediaType.TEXT_EVENT_STREAM)
        .retrieve()
        .bodyToFlux(type)
        .retryWhen(Retry.backoff(Long.MAX_VALUE, Duration.ofSeconds(1))
            .maxBackoff(Duration.ofSeconds(30))
            .doBeforeRetry(s -> log.warn("SSE disconnected, retrying: {}", s.failure().toString())))
        .filter(sse -> "order_updated".equals(sse.event()))
        .mapNotNull(ServerSentEvent::data)
        .filter(ev -> "SETTLED".equals(ev.status()))
        .doOnNext(this::handle)
        .subscribe();
  }

  private void handle(OrderEvent ev) {
    log.info("Settled {} at {} (tx={})", ev.intentId(), ev.settledAt(), ev.transactionHash());
    accounting.reconcile(ev.intentId(), ev.transactionHash(), ev.settledAt(),
                         ev.sourceAmountActual(), ev.targetAmount(), ev.fxRate());
  }
}