Assecutor/swyxweb
4
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Anrufer- und Webhook-Verarbeitung verbessern

Browse Source
This commit is contained in:
Sven committed 2026-11-02 15:58:53 +01:00
1 parent 7b3c187985
commit 9e71fe3a11
18 files changed
+1283 -111
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

backend/src/main/java/de/appcreation/swyxweb/storage/ArchiveService.java
+23 -18
View File
@@ -50,31 +50,36 @@ public class ArchiveService {
}
/**
* Legt Adressdaten-Einträge in der Sammlung {@code addresses} ab – alle
* Quellen (Webhooks Kunden/Kuriere, SwyxTray-Telefonbuch) in derselben
* Sammlung, je Eintrag nur Rufnummer, Name und Beschreibung. Die Sammlung
* ist ein Verzeichnis, kein Protokoll: Ein Eintrag zu einer bekannten
* Rufnummer wird aktualisiert statt verdoppelt – die SwyxTray-App schickt
* bei jeder Cache-Änderung das ganze Telefonbuch.
* Übernimmt vollständige Adressdaten aus den Webhooks. Ein erneuter
* Eintrag zur selben Rufnummer ersetzt auch die Kunden-/Job-Zuordnung.
*/
public void saveAddresses(List<AddressEntry> entries) {
saveAddresses(entries, true);
}
/** Der Telefonbuch-Cache kennt keine Kundenzuordnungen und darf sie nicht löschen. */
public void savePhonebookAddresses(List<AddressEntry> entries) {
saveAddresses(entries, false);
}
private void saveAddresses(List<AddressEntry> entries, boolean replaceAssignments) {
if (entries.isEmpty()) return;
// Upsert über die Rufnummer: aktualisiert den bekannten Eintrag oder
// legt ihn an; die Dokument-Kennung vergibt dabei die MongoDB.
submit("Adressdaten", () -> entries.forEach(entry -> mongo.upsert(
new Query(Criteria.where("number").is(entry.number())),
new Update()
.set("name", entry.name())
.set("description", entry.description())
.set("role", entry.role())
submit("Adressdaten", () -> entries.forEach(entry -> {
Update update = new Update()
.set("name", entry.name())
.set("description", entry.description())
.unset("receivedAt")
.unset("source");
if (replaceAssignments) {
update.set("role", entry.role())
.set("cscId", entry.cscId())
.set("jobIds", entry.jobIds())
.set("url", entry.url())
// Altbestand angleichen: Diese Felder wurden früher
// mitgeschrieben und sollen aus der Sammlung verschwinden.
.unset("receivedAt")
.unset("source"),
AddressEntry.class)));
.set("url", entry.url());
}
mongo.upsert(new Query(Criteria.where("number").is(entry.number())), update, AddressEntry.class);
}));
}
/**
backend/src/main/java/de/appcreation/swyxweb/storage/JobEntry.java
+4
View File
@@ -143,6 +143,10 @@ public record JobEntry(
/** Füllt die Klasse aus der Webhook-Nutzlast und stempelt die Eingangszeit. */
public static JobEntry from(JsonNode json, ObjectMapper mapper, Instant receivedAt) {
if (json == null || !json.isObject() || !json.path("id").isIntegralNumber()
|| !json.path("id").canConvertToLong() || json.path("id").longValue() <= 0) {
throw new IllegalArgumentException("Ein Job benötigt eine positive ganzzahlige id im Long-Wertebereich.");
}
JobEntry job = mapper.convertValue(json, JobEntry.class);
return new JobEntry(job.id(), receivedAt, job.url(), job.mapUrl(), job.state(), job.ordertime(),
job.orderdate(), job.modified(), job.finished(), job.vehicle(), job.service(),
backend/src/main/java/de/appcreation/swyxweb/web/AddressController.java
+3 -2
View File
@@ -24,7 +24,8 @@ import de.appcreation.swyxweb.storage.ArchiveService;
* <p>{@code POST} nimmt die Adress-Nachricht der SwyxTray-App entgegen – die
* WebSocket-Verbindung zur App hält nur der Browser, deshalb meldet die
* Startseite den Adress-Cache hierher. Angenommen wird jede Form, die
* {@link AddressEntry#manyFrom} versteht; die Adressdaten fremder Systeme
* {@link AddressEntry#manyFrom} versteht. Aktualisiert werden nur Name und
* Beschreibung, bestehende Kundenzuordnungen bleiben erhalten. Die Adressdaten fremder Systeme
* kommen dagegen normalerweise über {@code POST /api/webhook} herein.
*
* <p>{@code GET} liefert die gespeicherten Einträge aus der MongoDB zurück,
@@ -53,7 +54,7 @@ public class AddressController {
if (entries.isEmpty()) {
throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "Keine Adressdaten in der Nutzlast.");
}
archive.saveAddresses(entries);
archive.savePhonebookAddresses(entries);
return entries.size();
}
backend/src/main/java/de/appcreation/swyxweb/web/WebhookController.java
+4 -4
View File
@@ -47,10 +47,10 @@ import de.appcreation.swyxweb.webhook.WebhookService;
* -d '{"id":4711,"state":1,"customer":{"csc_id":100164,"name":"SYSGEN GmbH"},"tours":[…]}'</pre>
*
* <p>Daneben bleibt der generische {@code POST /api/webhook}: Er nimmt
* <b>beliebiges</b> JSON an – {@code "type":"job"} wird als Job abgelegt, alles
* andere als Adressdaten. Die Startseite zeigt jede Nachricht im Bereich
* „Webhook" unverändert an; das aufrufende System muss sich an kein festes
* Schema halten.
* beliebiges JSON an. Nachrichten mit {@code "type":"job"} müssen wie am
* Job-Endpunkt ein gültiger Job mit positiver ganzzahliger {@code id} sein;
* ungültige Jobdaten ergeben HTTP 400. Andere Nachrichten brauchen kein
* festes Schema. Die Startseite zeigt angenommene Nutzlasten unverändert an.
*
* <p>Der Weg zum Browser läuft über {@code GET /api/webhook/events}
* (Server-Sent Events), siehe {@link WebhookService}.
backend/src/main/java/de/appcreation/swyxweb/webhook/WebhookService.java
+35 -49
View File
@@ -3,7 +3,6 @@ package de.appcreation.swyxweb.webhook;
import java.io.IOException;
import java.time.Instant;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Deque;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
@@ -16,10 +15,13 @@ import jakarta.annotation.PreDestroy;
import tools.jackson.databind.JsonNode;
import tools.jackson.databind.ObjectMapper;
import tools.jackson.core.JacksonException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import org.springframework.http.HttpStatus;
import org.springframework.web.server.ResponseStatusException;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import de.appcreation.swyxweb.config.WebhookProperties;
@@ -110,46 +112,34 @@ public class WebhookService {
* @return das angelegte Ereignis, damit der Aufrufer die Id quittieren kann
*/
public WebhookEvent record(JsonNode payload, String channel) {
WebhookEvent event = new WebhookEvent(nextId.getAndIncrement(), Instant.now(), channel, payload);
Instant receivedAt = Instant.now();
boolean isJob = "jobs".equals(channel) || (channel == null && JobEntry.isJob(payload));
// Ungültige Jobs dürfen weder quittiert noch an Browser verteilt werden.
JobEntry job = isJob ? parseJob(payload, receivedAt) : null;
List<AddressEntry> addresses = isJob ? List.of() : AddressEntry.manyFrom(payload);
synchronized (events) {
// Auch Id-Vergabe und Versand gehören zum Monitor: parallele POSTs
// und eine SSE-Nachlieferung müssen dieselbe Reihenfolge sehen.
WebhookEvent event = new WebhookEvent(nextId.getAndIncrement(), receivedAt, channel, payload);
if (job != null) archive.saveJob(job);
else archive.saveAddresses(addresses);
events.addLast(event);
while (events.size() > Math.max(1, properties.history())) {
events.removeFirst();
}
log.debug("Webhook: Nachricht {} angenommen ({} Empfänger).", event.id(), subscribers.size());
broadcast(event);
return event;
}
log.debug("Webhook: Nachricht {} angenommen ({} Empfänger).", event.id(), subscribers.size());
// Den Inhalt dauerhaft in die MongoDB – nebenbei, die Quittung wartet
// nicht darauf.
try {
archiveByChannel(event, channel);
} catch (RuntimeException e) {
// Eine unerwartet geformte Nutzlast darf die Quittung nicht kippen.
log.warn("Webhook: Nachricht {} nicht in die MongoDB übernommen ({}).", event.id(), e.getMessage());
}
broadcast(event);
return event;
}
/**
* Die URL bestimmt die Ablage: Kunden und Kuriere sind Adressdaten, Jobs
* sind Jobs. Beim generischen Webhook ({@code channel == null}) entscheidet
* die Nutzlast selbst: {@code "type":"job"} ist ein Job, alles andere sind
* Adressdaten. Alle Adressdaten landen in derselben Sammlung.
*/
private void archiveByChannel(WebhookEvent event, String channel) {
JsonNode payload = event.payload();
switch (channel == null ? "" : channel) {
case "kunden", "kuriere" -> archive.saveAddresses(AddressEntry.manyFrom(payload));
case "jobs" -> archive.saveJob(JobEntry.from(payload, mapper, event.receivedAt()));
default -> {
if (JobEntry.isJob(payload)) {
archive.saveJob(JobEntry.from(payload, mapper, event.receivedAt()));
} else {
archive.saveAddresses(AddressEntry.manyFrom(payload));
}
}
private JobEntry parseJob(JsonNode payload, Instant receivedAt) {
try {
return JobEntry.from(payload, mapper, receivedAt);
} catch (JacksonException | IllegalArgumentException e) {
throw new ResponseStatusException(HttpStatus.BAD_REQUEST,
"Ungültige Jobdaten: Ein Job benötigt eine positive ganzzahlige id und passende Feldtypen.", e);
}
}
@@ -186,25 +176,21 @@ public class WebhookService {
});
emitter.onError(e -> subscribers.remove(emitter));
List<WebhookEvent> missed = new ArrayList<>();
if (lastEventId != null) {
for (WebhookEvent event : history()) {
if (event.id() > lastEventId) missed.add(event);
synchronized (events) {
try {
// Nachlieferung und Anmeldung sind atomar gegenüber record():
// Ein neuer Eintrag kommt entweder im Replay oder live an.
emitter.send(SseEmitter.event().comment("verbunden"));
if (lastEventId != null) {
for (WebhookEvent event : events) {
if (event.id() > lastEventId) emitter.send(toSse(event));
}
}
subscribers.add(emitter);
} catch (IOException | IllegalStateException e) {
emitter.completeWithError(e);
}
}
try {
// Ohne erste Nachricht bliebe die Antwort ohne Kopfzeilen hängen.
emitter.send(SseEmitter.event().comment("verbunden"));
for (WebhookEvent event : missed) {
emitter.send(toSse(event));
}
} catch (IOException | IllegalStateException e) {
emitter.completeWithError(e);
return emitter;
}
subscribers.add(emitter);
return emitter;
}
backend/src/test/java/de/appcreation/swyxweb/AddressControllerTests.java
+1 -1
View File
@@ -47,7 +47,7 @@ class AddressControllerTests {
.andExpect(status().isAccepted())
.andExpect(jsonPath("$").value(2));
verify(archive).saveAddresses(argThat(entries -> entries.size() == 2));
verify(archive).savePhonebookAddresses(argThat(entries -> entries.size() == 2));
}
/** Die Suche läuft in der Datenbank – der Begriff wird durchgereicht. */
backend/src/test/java/de/appcreation/swyxweb/WebhookControllerTests.java
+33
View File
@@ -3,6 +3,7 @@ package de.appcreation.swyxweb;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.argThat;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoInteractions;
import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.delete;
import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
@@ -23,6 +24,8 @@ import org.springframework.http.MediaType;
import org.springframework.test.context.TestPropertySource;
import org.springframework.test.context.bean.override.mockito.MockitoBean;
import org.springframework.test.web.servlet.MockMvc;
import tools.jackson.databind.ObjectMapper;
import tools.jackson.databind.node.ObjectNode;
/**
* Der Webhook nimmt beliebiges JSON an und hält es im Verlauf vor. Ohne
@@ -38,6 +41,9 @@ class WebhookControllerTests {
@Autowired
WebhookService service;
@Autowired
ObjectMapper mapper;
/** Ersetzt: Die Tests laufen ohne MongoDB und hinterlassen dort nichts. */
@MockitoBean
ArchiveService archive;
@@ -151,6 +157,33 @@ class WebhookControllerTests {
&& "https://vhs.assecutor.org/?jb=21891233&cr=13385".equals(job.mapUrl())));
}
@ParameterizedTest
@ValueSource(strings = {"{}", "{\"id\":null}", "{\"id\":0}", "{\"id\":-1}",
"{\"id\":\"4711\"}", "{\"id\":1.5}", "{\"id\":9223372036854775808}",
"{\"id\":4711,\"customer\":[]}", "{\"id\":4711,\"tours\":\"ungueltig\"}"})
void rejectsInvalidJobsBeforeRecordingOrArchiving(String payload) throws Exception {
mvc.perform(post("/api/webhook/jobs").contentType(MediaType.APPLICATION_JSON).content(payload))
.andExpect(status().isBadRequest());
ObjectNode genericJob = (ObjectNode) mapper.readTree(payload);
genericJob.put("type", "job");
mvc.perform(post("/api/webhook").contentType(MediaType.APPLICATION_JSON)
.content(mapper.writeValueAsString(genericJob)))
.andExpect(status().isBadRequest());
assertThat(service.history()).isEmpty();
verifyNoInteractions(archive);
}
@ParameterizedTest
@ValueSource(strings = {"[]", "[ {\"id\":4711} ]", "42", "\"job\""})
void jobEndpointRequiresOneObject(String payload) throws Exception {
mvc.perform(post("/api/webhook/jobs").contentType(MediaType.APPLICATION_JSON).content(payload))
.andExpect(status().isBadRequest());
assertThat(service.history()).isEmpty();
verifyNoInteractions(archive);
}
/** Ohne Token in der Konfiguration darf der Kopf fehlen. */
@Test
void tokenIsNotRequiredWhenUnset() throws Exception {
backend/src/test/java/de/appcreation/swyxweb/storage/ArchiveServiceTests.java
+60
View File
@@ -0,0 +1,60 @@
package de.appcreation.swyxweb.storage;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.*;
import java.util.List;
import org.bson.Document;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.UpdateDefinition;
class ArchiveServiceTests {
private final MongoTemplate mongo = mock(MongoTemplate.class);
private final ArchiveService archive = new ArchiveService(mongo);
@AfterEach
void shutdown() {
archive.shutdown();
}
@Test
void phonebookRefreshPreservesPreviouslyImportedCustomerAssignments() {
archive.saveAddresses(List.of(new AddressEntry(null, "+493012345", "Kunde", "CRM",
"customer", 42, List.of(4711L), "https://example.test/customer/42")));
archive.savePhonebookAddresses(List.of(new AddressEntry(null, "+493012345", "Telefonbuch", "Swyx",
null, null, null, null)));
var queries = ArgumentCaptor.forClass(Query.class);
var updates = ArgumentCaptor.forClass(UpdateDefinition.class);
verify(mongo, timeout(2000).times(2)).upsert(queries.capture(), updates.capture(), eq(AddressEntry.class));
assertThat(queries.getAllValues()).allSatisfy(query ->
assertThat(query.getQueryObject()).containsEntry("number", "+493012345"));
Document customer = updates.getAllValues().getFirst().getUpdateObject().get("$set", Document.class);
assertThat(customer).containsEntry("role", "customer").containsEntry("cscId", 42)
.containsEntry("jobIds", List.of(4711L)).containsEntry("url", "https://example.test/customer/42");
Document phonebookUpdate = updates.getAllValues().getLast().getUpdateObject();
assertThat(phonebookUpdate.get("$set", Document.class))
.containsEntry("name", "Telefonbuch").containsEntry("description", "Swyx");
// Kein Mongo-Operator darf die bestehenden Kundenzuordnungen verändern.
phonebookUpdate.values().forEach(fields -> assertThat((Document) fields)
.doesNotContainKeys("role", "cscId", "jobIds", "url"));
}
@Test
void completeWebhookUpdateCanStillRemoveAssignments() {
archive.saveAddresses(List.of(new AddressEntry(null, "+493012345", "Kunde", null,
"customer", 42, List.of(), null)));
var update = ArgumentCaptor.forClass(UpdateDefinition.class);
verify(mongo, timeout(2000)).upsert(any(Query.class), update.capture(), eq(AddressEntry.class));
assertThat(update.getValue().getUpdateObject().get("$set", Document.class))
.containsEntry("jobIds", List.of()).containsEntry("url", null);
}
}
backend/src/test/java/de/appcreation/swyxweb/webhook/WebhookStreamTests.java
+103
View File
@@ -0,0 +1,103 @@
package de.appcreation.swyxweb.webhook;
import static org.assertj.core.api.Assertions.*;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.*;
import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
import org.springframework.mock.web.MockHttpServletResponse;
import org.springframework.test.web.servlet.MockMvc;
import org.springframework.test.web.servlet.setup.MockMvcBuilders;
import de.appcreation.swyxweb.config.WebhookProperties;
import de.appcreation.swyxweb.storage.ArchiveService;
import de.appcreation.swyxweb.web.WebhookController;
import tools.jackson.databind.ObjectMapper;
class WebhookStreamTests {
private final ObjectMapper mapper = spy(new ObjectMapper());
private final WebhookProperties properties = new WebhookProperties("", 50, 262144);
private final WebhookService service = new WebhookService(properties, mapper, mock(ArchiveService.class));
private final MockMvc mvc = MockMvcBuilders.standaloneSetup(new WebhookController(service, properties, mapper)).build();
private final ExecutorService threads = Executors.newFixedThreadPool(4);
@AfterEach
void shutdown() {
threads.shutdownNow();
service.shutdown();
}
@Test
void concurrentPostDuringReplayIsDeliveredExactlyOnceAfterReplay() throws Exception {
WebhookEvent first = service.record(mapper.readTree("{\"n\":1}"));
var replayStarted = new CountDownLatch(1);
var releaseReplay = new CountDownLatch(1);
doAnswer(invocation -> {
WebhookEvent event = invocation.getArgument(0);
if (event.id() == first.id()) {
replayStarted.countDown();
if (!releaseReplay.await(5, TimeUnit.SECONDS)) throw new AssertionError("Replay was not released");
}
return invocation.callRealMethod();
}).when(mapper).writeValueAsString(any(WebhookEvent.class));
Future<MockHttpServletResponse> subscription = threads.submit(() -> subscribe("0"));
try {
assertThat(replayStarted.await(5, TimeUnit.SECONDS)).isTrue();
var postStarted = new CountDownLatch(1);
Future<WebhookEvent> post = threads.submit(() -> {
postStarted.countDown();
return service.record(mapper.readTree("{\"n\":2}"));
});
assertThat(postStarted.await(5, TimeUnit.SECONDS)).isTrue();
// Der POST darf die Lücke zwischen Replay und Anmeldung nicht passieren.
assertThatThrownBy(() -> post.get(150, TimeUnit.MILLISECONDS)).isInstanceOf(TimeoutException.class);
releaseReplay.countDown();
MockHttpServletResponse response = subscription.get(5, TimeUnit.SECONDS);
WebhookEvent second = post.get(5, TimeUnit.SECONDS);
WebhookEvent third = service.record(mapper.readTree("{\"n\":3}"));
assertThat(eventIds(response)).containsExactly(first.id(), second.id(), third.id());
} finally {
releaseReplay.countDown();
}
}
@Test
void reconnectReplaysOnlyMissedEventsAndThenStreamsNewOnes() throws Exception {
WebhookEvent first = service.record(mapper.readTree("{\"n\":1}"));
WebhookEvent second = service.record(mapper.readTree("{\"n\":2}"));
MockHttpServletResponse response = subscribe(Long.toString(first.id()));
WebhookEvent third = service.record(mapper.readTree("{\"n\":3}"));
assertThat(eventIds(response)).containsExactly(second.id(), third.id());
}
@Test
void parallelPostsKeepTheSameOrderInHistoryAndStream() throws Exception {
MockHttpServletResponse response = subscribe("0");
List<Future<WebhookEvent>> posts = new ArrayList<>();
for (int i = 0; i < 20; i++) {
posts.add(threads.submit(() -> service.record(mapper.readTree("{}"))));
}
for (Future<WebhookEvent> post : posts) post.get(5, TimeUnit.SECONDS);
List<Long> ids = eventIds(response);
assertThat(ids).hasSize(20).doesNotHaveDuplicates().isSorted();
assertThat(service.history()).extracting(WebhookEvent::id).containsExactlyElementsOf(ids);
}
private MockHttpServletResponse subscribe(String lastId) throws Exception {
return mvc.perform(get("/api/webhook/events").header("Last-Event-ID", lastId))
.andReturn().getResponse();
}
private List<Long> eventIds(MockHttpServletResponse response) throws Exception {
return response.getContentAsString().lines().filter(line -> line.startsWith("id:"))
.map(line -> Long.parseLong(line.substring(3).trim())).toList();
}
}