Apple Messages for Business

import hashlib
import hmac
import logging
import os

import httpx
from fastapi import BackgroundTasks, FastAPI, Header, HTTPException, Request

from docs_assistant.core import ask  # the channel-agnostic core (async generator of events)

app = FastAPI()
log = logging.getLogger("apple_webhook")

WEBHOOK_SECRET = os.environ["MSP_WEBHOOK_SECRET"]      # placeholder: set in your environment
MSP_API_URL = os.environ["MSP_API_URL"]                # placeholder: your MSP's send endpoint
MSP_API_TOKEN = os.environ["MSP_API_TOKEN"]            # placeholder: your MSP's credential

seen: set[str] = set()  # use Redis with a TTL in production


def verify(body: bytes, signature: str) -> bool:
    expected = hmac.new(WEBHOOK_SECRET.encode(), body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature)


async def answer_and_reply(msp_conversation_id: str, customer_id: str, text: str) -> None:
    parts: list[str] = []
    sources: list[str] = []
    # Prefixed keys are only for the assistant's memory; the MSP gets its own raw id back.
    async for event in ask(f"apple:{msp_conversation_id}", f"apple:{customer_id}", text):
        if event["type"] == "token":
            parts.append(event["text"])
        elif event["type"] == "citation":
            sources.append(event["title"])
    reply = "".join(parts) or "Sorry, I could not find an answer."
    if sources:
        reply += "\n\nSources: " + ", ".join(sources[:3])
    async with httpx.AsyncClient(timeout=10) as client:
        response = await client.post(
            MSP_API_URL,
            headers={"Authorization": f"Bearer {MSP_API_TOKEN}"},
            json={"conversationId": msp_conversation_id, "text": reply},
        )
    try:
        response.raise_for_status()
    except httpx.HTTPStatusError:
        log.exception("MSP rejected reply for conversation %s", msp_conversation_id)
        raise


@app.post("/apple/webhook", status_code=202)
async def webhook(request: Request, background: BackgroundTasks,
                  x_signature: str = Header(...)):
    body = await request.body()
    if not verify(body, x_signature):
        raise HTTPException(status_code=401)
    msg = await request.json()
    if msg["messageId"] in seen:
        return {"status": "duplicate"}
    seen.add(msg["messageId"])
    background.add_task(
        answer_and_reply,
        msg["conversationId"],
        msg["customerId"],
        msg["text"],
    )
    return {"status": "accepted"}

import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.client.RestClient;

@RestController
class AppleMessagesWebhook {

    private static final Logger log = LoggerFactory.getLogger(AppleMessagesWebhook.class);

    record Inbound(String messageId, String conversationId, String customerId, String text) {}

    private final DocsAssistantCore core;
    private final RestClient msp;
    private final byte[] secret;
    private final com.fasterxml.jackson.databind.ObjectMapper mapper;
    private final Set<String> seen = ConcurrentHashMap.newKeySet(); // Redis with TTL in production
    private final ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();

    AppleMessagesWebhook(DocsAssistantCore core,
                         com.fasterxml.jackson.databind.ObjectMapper mapper,
                         RestClient.Builder builder,
                         @Value("${MSP_API_URL}") String mspUrl,          // placeholder
                         @Value("${MSP_API_TOKEN}") String mspToken,      // placeholder
                         @Value("${MSP_WEBHOOK_SECRET}") String secret) { // placeholder
        this.core = core;
        this.mapper = mapper;
        this.msp = builder.baseUrl(mspUrl)
                .defaultHeader("Authorization", "Bearer " + mspToken).build();
        this.secret = secret.getBytes(StandardCharsets.UTF_8);
    }

    @PostMapping("/apple/webhook")
    ResponseEntity<Map<String, String>> receive(@RequestBody byte[] body,
                                                @RequestHeader("X-Signature") String signature)
            throws Exception {
        if (!valid(body, signature)) {
            return ResponseEntity.status(HttpStatus.UNAUTHORIZED).build();
        }
        Inbound in = mapper.readValue(body, Inbound.class);
        if (!seen.add(in.messageId())) {
            return ResponseEntity.accepted().body(Map.of("status", "duplicate"));
        }
        CompletableFuture.runAsync(() -> {
            // Prefixed keys are only for the assistant's memory; the MSP gets its own raw id back.
            String reply = core.answer("apple:" + in.conversationId(), "apple:" + in.customerId(), in.text());
            msp.post().body(Map.of("conversationId", in.conversationId(), "text", reply))
               .retrieve().toBodilessEntity();
        }, executor).exceptionally(e -> {
            log.error("Failed to answer conversation {}", in.conversationId(), e);
            return null;
        });
        return ResponseEntity.accepted().body(Map.of("status", "accepted"));
    }

    private boolean valid(byte[] body, String signature) throws Exception {
        Mac mac = Mac.getInstance("HmacSHA256");
        mac.init(new SecretKeySpec(secret, "HmacSHA256"));
        String expected = java.util.HexFormat.of().formatHex(mac.doFinal(body));
        return MessageDigest.isEqual(expected.getBytes(StandardCharsets.UTF_8),
                                     signature.getBytes(StandardCharsets.UTF_8));
    }
}