package com.homme.demo.service; import java.time.LocalDate; import java.util.ArrayList; import java.util.List; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.http.HttpStatus; import org.springframework.http.ResponseEntity; import org.springframework.stereotype.Service; import com.homme.demo.entity.Document; import com.homme.demo.entity.DocumentChunck; import com.homme.demo.repository.DocumentChunckRepository; import com.homme.demo.repository.DocumentRepository; import jakarta.transaction.Transactional; @Service public class DocumentStorageService { @Autowired private DocumentRepository documentRepossitory; @Autowired private DocumentChunckRepository documentChunckRepository; @Autowired private DocumentChunckService documentChunkService; @Autowired private AzureEmbeddingService azureEmbeddingService; @Transactional public ResponseEntity saveDocuemntWithChuncks( String sourceUrl, String rawText, String cleanText) { String message = ""; if(documentRepossitory.existsBySourceUrl(sourceUrl)) { message = "Dieses Dokument wurde bereits gespeichert."; return new ResponseEntity<>(message, HttpStatus.CONFLICT); } Document document = Document.builder() .sourceUrl(sourceUrl) .rawText(rawText) .cleanText(cleanText) .createdAt(LocalDate.now()) .build(); Integer savedDocumentId = documentRepossitory.save(document).getId(); ExecutorService pool = Executors.newFixedThreadPool(4); try { List chuncks = documentChunkService.splitTextIntoChunks(cleanText); List> futures = new ArrayList<>(); for(int i = 0; i < chuncks.size(); i++) { String chunckText = chuncks.get(i); futures.add(pool.submit(() -> (chunckText == null || chunckText.isBlank()) ?null : azureEmbeddingService.createEmbeddingFloatArray(chunckText, "document") )); } for(int i=0; i < chuncks.size(); i++) { float [] embedding = futures.get(i).get(); String chunckText = chuncks.get(i); DocumentChunck documentChunck = DocumentChunck.builder() .chunckIndex(i) .chunckText(chunckText) .documentId(savedDocumentId) .embedding(embedding) .build(); documentChunckRepository.save(documentChunck); } message = "Dokument wurde erfolgreich gespeichert."; return new ResponseEntity<>(message, HttpStatus.CREATED); }catch(InterruptedException | ExecutionException e) { Thread.currentThread().interrupt(); message = "Fehler bei parallerer Verarbeitung"+ e.getMessage(); return new ResponseEntity<>(message, HttpStatus.BAD_REQUEST); } finally { pool.shutdown(); } } @Transactional public int fillMissingEmbedding() { List chuncks = documentChunckRepository.findByEmbeddingIsNull(); ExecutorService pool = Executors.newFixedThreadPool(6); List> futures = new ArrayList<>(); try { for(DocumentChunck chunck: chuncks) { String chunckText = chunck.getChunckText(); futures.add(pool.submit(() -> (chunckText == null || chunckText.isBlank()) ? null : azureEmbeddingService.createEmbeddingFloatArray(chunckText) )); } int updatedCount = 0; for( int i = 0; i < chuncks.size(); i++) { float[] embedding = futures.get(i).get(); chuncks.get(i).setEmbedding(embedding); documentChunckRepository.save(chuncks.get(i)); updatedCount++; System.out.println("updatedCount = "+updatedCount); } return updatedCount; }catch(InterruptedException | ExecutionException e ) { Thread.currentThread().interrupt(); throw new RuntimeException("Fehler bei parallerer Verarbeitung: "+e.getMessage()); }finally { pool.shutdown(); } } }