From 10c04eaf67b21826676e52be58d64024d5b97ef7 Mon Sep 17 00:00:00 2001 From: let5sne Date: Sun, 12 Jul 2026 03:14:57 +0800 Subject: [PATCH] fix(personal): isolate personal objects and dedupe under lock --- .../service/PersonalIngestionService.java | 312 +++++++++-------- .../service/PersonalIngestionWorker.java | 15 +- .../PersonalIngestionServiceTest.java | 322 ++++++++++-------- .../personal/PersonalIngestionWorkerTest.java | 14 +- 4 files changed, 363 insertions(+), 300 deletions(-) diff --git a/backend/ruoyi-modules/ruoyi-aihr/src/main/java/org/dromara/aihr/personal/service/PersonalIngestionService.java b/backend/ruoyi-modules/ruoyi-aihr/src/main/java/org/dromara/aihr/personal/service/PersonalIngestionService.java index 6268b97a..e98e6390 100644 --- a/backend/ruoyi-modules/ruoyi-aihr/src/main/java/org/dromara/aihr/personal/service/PersonalIngestionService.java +++ b/backend/ruoyi-modules/ruoyi-aihr/src/main/java/org/dromara/aihr/personal/service/PersonalIngestionService.java @@ -1,5 +1,6 @@ package org.dromara.aihr.personal.service; +import com.baomidou.mybatisplus.core.toolkit.IdWorker; import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; import lombok.extern.slf4j.Slf4j; @@ -8,12 +9,13 @@ import org.dromara.aihr.personal.domain.PersonalAssistantDto.TextItemRequest; import org.dromara.aihr.personal.support.PersonalKnowledgeProperties; import org.dromara.aihr.personal.support.PersonalOwner; import org.dromara.common.core.exception.ServiceException; +import org.dromara.common.oss.core.OssClient; +import org.dromara.common.oss.entity.UploadResult; import org.dromara.common.oss.factory.OssFactory; import org.dromara.system.domain.vo.SysOssVo; import org.dromara.system.service.ISysOssService; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.jdbc.core.JdbcTemplate; -import org.springframework.jdbc.support.GeneratedKeyHolder; -import org.springframework.jdbc.support.KeyHolder; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.support.TransactionSynchronization; @@ -21,16 +23,10 @@ import org.springframework.transaction.support.TransactionSynchronizationManager import org.springframework.web.multipart.MultipartFile; import java.io.ByteArrayInputStream; -import java.io.File; import java.io.IOException; -import java.io.InputStream; import java.nio.charset.StandardCharsets; -import java.nio.file.Files; -import java.nio.file.Path; import java.security.MessageDigest; import java.security.NoSuchAlgorithmException; -import java.sql.PreparedStatement; -import java.sql.Statement; import java.time.LocalDateTime; import java.util.HexFormat; import java.util.List; @@ -38,12 +34,15 @@ import java.util.Locale; import java.util.Map; import java.util.Set; import java.util.UUID; +import java.util.function.LongSupplier; +import java.util.regex.Pattern; @Slf4j @Service public class PersonalIngestionService { private static final String ITEM_NOT_FOUND = "PERSONAL_ITEM_NOT_FOUND"; + private static final Pattern SAFE_TENANT = Pattern.compile("[A-Za-z0-9_-]{1,20}"); private static final Set SUPPORTED_FILE_SUFFIXES = Set.of( "txt", "md", "markdown", "pdf", "doc", "docx", "xls", "xlsx", "ppt", "pptx" ); @@ -53,55 +52,54 @@ public class PersonalIngestionService { private final PersonalKnowledgeProperties properties; private final ISysOssService ossService; private final ObjectMapper objectMapper; + private final PersonalObjectStore objectStore; + private final LongSupplier itemIdSupplier; + @Autowired public PersonalIngestionService(JdbcTemplate jdbcTemplate, PersonalSpaceService spaceService, PersonalKnowledgeProperties properties, ISysOssService ossService, ObjectMapper objectMapper) { + this(jdbcTemplate, spaceService, properties, ossService, objectMapper, + new DefaultPersonalObjectStore(jdbcTemplate), IdWorker::getId); + } + + private PersonalIngestionService(JdbcTemplate jdbcTemplate, PersonalSpaceService spaceService, + PersonalKnowledgeProperties properties, ISysOssService ossService, + ObjectMapper objectMapper, PersonalObjectStore objectStore, + LongSupplier itemIdSupplier) { this.jdbcTemplate = jdbcTemplate; this.spaceService = spaceService; this.properties = properties; this.ossService = ossService; this.objectMapper = objectMapper; + this.objectStore = objectStore; + this.itemIdSupplier = itemIdSupplier; + } + + public static PersonalIngestionService forTest(JdbcTemplate jdbcTemplate, PersonalSpaceService spaceService, + PersonalKnowledgeProperties properties, ISysOssService ossService, + ObjectMapper objectMapper, PersonalObjectStore objectStore, + LongSupplier itemIdSupplier) { + return new PersonalIngestionService(jdbcTemplate, spaceService, properties, ossService, objectMapper, + objectStore, itemIdSupplier); } @Transactional public ItemCreatedResponse createText(PersonalOwner owner, TextItemRequest request) { + validateOwner(owner); if (request == null || request.content() == null || request.content().isBlank()) { throw new ServiceException("PERSONAL_TEXT_EMPTY"); } byte[] bytes = request.content().getBytes(StandardCharsets.UTF_8); validateSize(bytes.length); - String hash = sha256(bytes); - ItemCreatedResponse duplicate = duplicate(owner, hash); - if (duplicate != null) { - return duplicate; - } - - long spaceId = spaceService.reserve(owner, bytes.length); - Path temporary = null; - SysOssVo uploaded; - try { - temporary = Files.createTempFile("personal-" + UUID.randomUUID(), ".txt"); - Files.write(temporary, bytes); - uploaded = ossService.upload(temporary.toFile()); - } catch (IOException ex) { - throw new ServiceException("PERSONAL_TEXT_STORAGE_FAILED"); - } finally { - if (temporary != null) { - try { - Files.deleteIfExists(temporary); - } catch (IOException ex) { - log.warn("Unable to remove temporary personal text object"); - } - } - } - return persist(owner, spaceId, "TEXT", cleanTitle(request.title(), "文字资料"), - "text/plain", bytes.length, hash, request.capturedAt(), request.tags(), uploaded); + return create(owner, "TEXT", cleanTitle(request.title(), "文字资料"), "txt", "text/plain", bytes, + request.capturedAt(), request.tags()); } @Transactional public ItemCreatedResponse createFile(PersonalOwner owner, MultipartFile file, String title, LocalDateTime capturedAt) { + validateOwner(owner); validateFile(file); byte[] bytes; try { @@ -109,23 +107,14 @@ public class PersonalIngestionService { } catch (IOException ex) { throw new ServiceException("PERSONAL_FILE_READ_FAILED"); } - String hash = sha256(bytes); - ItemCreatedResponse duplicate = duplicate(owner, hash); - if (duplicate != null) { - return duplicate; - } - validateSize(bytes.length); - long spaceId = spaceService.reserve(owner, bytes.length); - String safeUploadName = "personal-" + UUID.randomUUID() + suffixWithDot(file.getOriginalFilename()); - SysOssVo uploaded = ossService.upload(new SafeMultipartFile( - safeUploadName, cleanMime(file.getContentType()), bytes)); String originalName = safeFileName(file.getOriginalFilename()); - return persist(owner, spaceId, "FILE", cleanTitle(title, originalName), - cleanMime(file.getContentType()), bytes.length, hash, capturedAt, List.of(), uploaded); + return create(owner, "FILE", cleanTitle(title, originalName), suffix(originalName), + cleanMime(file.getContentType()), bytes, capturedAt, List.of()); } public void retry(PersonalOwner owner, long itemId) { + validateOwner(owner); int updated = jdbcTemplate.update(""" update aihr_personal_item set status = 'QUEUED', error_code = null, error_message = null, @@ -138,37 +127,50 @@ public class PersonalIngestionService { } } - private ItemCreatedResponse persist(PersonalOwner owner, long spaceId, String sourceType, String title, - String mimeType, long size, String hash, LocalDateTime capturedAt, - List tags, SysOssVo uploaded) { + private ItemCreatedResponse create(PersonalOwner owner, String sourceType, String title, String suffix, + String mimeType, byte[] bytes, LocalDateTime capturedAt, List tags) { + String hash = sha256(bytes); + // reserve locks the current owner's space row. Dedupe must happen while that lock is held. + long spaceId = spaceService.reserve(owner, bytes.length); + ItemCreatedResponse duplicate = duplicate(owner, spaceId, hash); + if (duplicate != null) { + return duplicate; + } + + long itemId = positiveId(itemIdSupplier.getAsLong()); + String objectKey = objectKey(owner, itemId, suffix); + SysOssVo uploaded = objectStore.upload(owner, itemId, objectKey, suffix, mimeType, bytes); + return persist(owner, spaceId, itemId, sourceType, title, mimeType, bytes.length, hash, capturedAt, + tags, objectKey, uploaded); + } + + private ItemCreatedResponse persist(PersonalOwner owner, long spaceId, long itemId, String sourceType, + String title, String mimeType, long size, String hash, + LocalDateTime capturedAt, List tags, String objectKey, + SysOssVo uploaded) { if (uploaded == null || uploaded.getOssId() == null) { throw new ServiceException("PERSONAL_OSS_UPLOAD_FAILED"); } boolean deferredCleanup = registerRollbackCleanup(uploaded); try { - KeyHolder keyHolder = new GeneratedKeyHolder(); - jdbcTemplate.update(connection -> { - PreparedStatement statement = connection.prepareStatement(""" - insert into aihr_personal_item - (tenant_id, space_id, owner_user_id, source_type, title, oss_id, mime_type, - size_bytes, content_hash, status, tags_json, captured_at, create_time, update_time) - values (?, ?, ?, ?, ?, ?, ?, ?, ?, 'QUEUED', ?, ?, now(), now()) - """, Statement.RETURN_GENERATED_KEYS); - statement.setString(1, owner.tenantId()); - statement.setLong(2, spaceId); - statement.setLong(3, owner.userId()); - statement.setString(4, sourceType); - statement.setString(5, title); - statement.setLong(6, uploaded.getOssId()); - statement.setString(7, mimeType); - statement.setLong(8, size); - statement.setString(9, hash); - statement.setString(10, tagsJson(tags)); - statement.setObject(11, capturedAt == null ? LocalDateTime.now() : capturedAt); - return statement; - }, keyHolder); - Number key = keyHolder.getKey(); - if (key == null) { + int bound = jdbcTemplate.update(""" + update sys_oss + set ext1 = ?, update_time = now(), update_by = ? + where tenant_id = ? and oss_id = ? and create_by = ? and file_name = ? + """, personalOssExt(itemId), owner.userId(), owner.tenantId(), uploaded.getOssId(), + owner.userId(), objectKey); + if (bound != 1) { + throw new ServiceException("PERSONAL_OSS_BIND_FAILED"); + } + int inserted = jdbcTemplate.update(""" + insert into aihr_personal_item + (id, tenant_id, space_id, owner_user_id, source_type, title, oss_id, mime_type, + size_bytes, content_hash, status, tags_json, captured_at, create_time, update_time) + values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'QUEUED', ?, ?, now(), now()) + """, itemId, owner.tenantId(), spaceId, owner.userId(), sourceType, title, + uploaded.getOssId(), mimeType, size, hash, tagsJson(tags), + capturedAt == null ? LocalDateTime.now() : capturedAt); + if (inserted != 1) { throw new ServiceException("PERSONAL_ITEM_CREATE_FAILED"); } int counterUpdated = jdbcTemplate.update(""" @@ -179,24 +181,24 @@ public class PersonalIngestionService { if (counterUpdated != 1) { throw new ServiceException("PERSONAL_SPACE_NOT_AVAILABLE"); } - return new ItemCreatedResponse(key.longValue(), "QUEUED", null); + return new ItemCreatedResponse(itemId, "QUEUED", null); } catch (RuntimeException ex) { if (!deferredCleanup) { - cleanupOss(uploaded.getOssId()); + cleanupOss(uploaded); } throw ex; } } - private ItemCreatedResponse duplicate(PersonalOwner owner, String hash) { + private ItemCreatedResponse duplicate(PersonalOwner owner, long spaceId, String hash) { List> rows = jdbcTemplate.queryForList(""" select id, status from aihr_personal_item - where tenant_id = ? and owner_user_id = ? and content_hash = ? + where tenant_id = ? and owner_user_id = ? and space_id = ? and content_hash = ? and status <> 'DELETED' order by id desc limit 1 - """, owner.tenantId(), owner.userId(), hash); + """, owner.tenantId(), owner.userId(), spaceId, hash); if (rows.isEmpty()) { return null; } @@ -210,8 +212,7 @@ public class PersonalIngestionService { throw new ServiceException("PERSONAL_FILE_EMPTY"); } validateSize(file.getSize()); - String suffix = suffix(file.getOriginalFilename()); - if (!SUPPORTED_FILE_SUFFIXES.contains(suffix)) { + if (!SUPPORTED_FILE_SUFFIXES.contains(suffix(file.getOriginalFilename()))) { throw new ServiceException("PERSONAL_FILE_UNSUPPORTED"); } } @@ -236,32 +237,31 @@ public class PersonalIngestionService { @Override public void afterCompletion(int status) { if (status != STATUS_COMMITTED) { - cleanupRolledBackOss(uploaded); + cleanupOss(uploaded); } } }); return true; } - private void cleanupRolledBackOss(SysOssVo uploaded) { - if (uploaded.getService() != null && !uploaded.getService().isBlank() - && uploaded.getFileName() != null && !uploaded.getFileName().isBlank()) { - try { - // sys_oss participates in the outer transaction and may already be rolled back here, - // so remove the physical object by its private object key before metadata cleanup. - OssFactory.instance(uploaded.getService()).delete(uploaded.getFileName()); - } catch (RuntimeException cleanupError) { - log.warn("Unable to clean rolled-back personal OSS object id={}", uploaded.getOssId()); - } + private void cleanupOss(SysOssVo uploaded) { + try { + objectStore.deletePhysical(uploaded); + } catch (RuntimeException cleanupError) { + log.warn("Unable to clean personal OSS object id={}", uploaded.getOssId()); + } + try { + ossService.deleteWithValidByIds(List.of(uploaded.getOssId()), false); + } catch (RuntimeException cleanupError) { + log.warn("Unable to clean personal OSS metadata id={}", uploaded.getOssId()); } - cleanupOss(uploaded.getOssId()); } - private void cleanupOss(long ossId) { + private String personalOssExt(long itemId) { try { - ossService.deleteWithValidByIds(List.of(ossId), false); - } catch (RuntimeException cleanupError) { - log.warn("Unable to clean personal OSS object id={}", ossId); + return objectMapper.writeValueAsString(Map.of("source", "personal", "itemId", itemId)); + } catch (JsonProcessingException ex) { + throw new ServiceException("PERSONAL_OSS_BIND_FAILED"); } } @@ -280,6 +280,32 @@ public class PersonalIngestionService { } } + private static String objectKey(PersonalOwner owner, long itemId, String suffix) { + validateOwner(owner); + long safeItemId = positiveId(itemId); + String safeSuffix = suffix == null ? "" : suffix.toLowerCase(Locale.ROOT); + if (!SUPPORTED_FILE_SUFFIXES.contains(safeSuffix)) { + throw new ServiceException("PERSONAL_FILE_UNSUPPORTED"); + } + String randomName = UUID.randomUUID().toString().replace("-", ""); + return "personal/" + owner.tenantId() + "/" + owner.userId() + "/" + safeItemId + "/" + + randomName + "." + safeSuffix; + } + + private static void validateOwner(PersonalOwner owner) { + if (owner == null || owner.userId() <= 0 || owner.tenantId() == null + || !SAFE_TENANT.matcher(owner.tenantId()).matches()) { + throw new ServiceException("PERSONAL_OWNER_INVALID"); + } + } + + private static long positiveId(long id) { + if (id <= 0) { + throw new ServiceException("PERSONAL_ID_INVALID"); + } + return id; + } + private static String sha256(byte[] bytes) { try { return HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256").digest(bytes)); @@ -313,11 +339,6 @@ public class PersonalIngestionService { return dot < 0 ? "" : safe.substring(dot + 1).toLowerCase(Locale.ROOT); } - private static String suffixWithDot(String fileName) { - String suffix = suffix(fileName); - return suffix.isBlank() ? "" : "." + suffix; - } - private static String cleanMime(String value) { if (value == null || value.isBlank()) { return "application/octet-stream"; @@ -327,55 +348,62 @@ public class PersonalIngestionService { return separator < 0 ? mime : mime.substring(0, separator).trim(); } - private static final class SafeMultipartFile implements MultipartFile { - private final String fileName; - private final String contentType; - private final byte[] bytes; + public interface PersonalObjectStore { + SysOssVo upload(PersonalOwner owner, long itemId, String objectKey, String suffix, String mimeType, + byte[] bytes); - private SafeMultipartFile(String fileName, String contentType, byte[] bytes) { - this.fileName = fileName; - this.contentType = contentType; - this.bytes = bytes.clone(); + void deletePhysical(SysOssVo uploaded); + } + + private static final class DefaultPersonalObjectStore implements PersonalObjectStore { + private final JdbcTemplate jdbcTemplate; + + private DefaultPersonalObjectStore(JdbcTemplate jdbcTemplate) { + this.jdbcTemplate = jdbcTemplate; } @Override - public String getName() { - return "file"; + public SysOssVo upload(PersonalOwner owner, long itemId, String objectKey, String suffix, String mimeType, + byte[] bytes) { + OssClient storage = OssFactory.instance(); + UploadResult result = storage.upload( + new ByteArrayInputStream(bytes), objectKey, (long) bytes.length, mimeType); + long ossId = IdWorker.getId(); + String safeName = objectKey.substring(objectKey.lastIndexOf('/') + 1); + try { + int inserted = jdbcTemplate.update(""" + insert into sys_oss + (oss_id, tenant_id, file_name, original_name, file_suffix, url, ext1, + create_time, create_by, update_time, update_by, service) + values (?, ?, ?, ?, ?, ?, null, now(), ?, now(), ?, ?) + """, ossId, owner.tenantId(), result.getFilename(), safeName, "." + suffix, + result.getUrl(), owner.userId(), owner.userId(), storage.getConfigKey()); + if (inserted != 1) { + throw new ServiceException("PERSONAL_OSS_METADATA_FAILED"); + } + } catch (RuntimeException ex) { + try { + storage.delete(result.getFilename()); + } catch (RuntimeException cleanupError) { + log.warn("Unable to clean personal object after metadata failure"); + } + throw ex; + } + SysOssVo uploaded = new SysOssVo(); + uploaded.setOssId(ossId); + uploaded.setFileName(result.getFilename()); + uploaded.setOriginalName(safeName); + uploaded.setFileSuffix("." + suffix); + uploaded.setUrl(result.getUrl()); + uploaded.setService(storage.getConfigKey()); + return uploaded; } @Override - public String getOriginalFilename() { - return fileName; - } - - @Override - public String getContentType() { - return contentType; - } - - @Override - public boolean isEmpty() { - return bytes.length == 0; - } - - @Override - public long getSize() { - return bytes.length; - } - - @Override - public byte[] getBytes() { - return bytes.clone(); - } - - @Override - public InputStream getInputStream() { - return new ByteArrayInputStream(bytes); - } - - @Override - public void transferTo(File destination) throws IOException { - Files.write(destination.toPath(), bytes); + public void deletePhysical(SysOssVo uploaded) { + if (uploaded.getService() != null && uploaded.getFileName() != null) { + OssFactory.instance(uploaded.getService()).delete(uploaded.getFileName()); + } } } } diff --git a/backend/ruoyi-modules/ruoyi-aihr/src/main/java/org/dromara/aihr/personal/service/PersonalIngestionWorker.java b/backend/ruoyi-modules/ruoyi-aihr/src/main/java/org/dromara/aihr/personal/service/PersonalIngestionWorker.java index 0cf73f9f..b7ec3666 100644 --- a/backend/ruoyi-modules/ruoyi-aihr/src/main/java/org/dromara/aihr/personal/service/PersonalIngestionWorker.java +++ b/backend/ruoyi-modules/ruoyi-aihr/src/main/java/org/dromara/aihr/personal/service/PersonalIngestionWorker.java @@ -90,7 +90,8 @@ public class PersonalIngestionWorker { } try { - StoredObject stored = objectReader.read(item.ossId(), maxInputBytes); + StoredObject stored = objectReader.read( + item.ossId(), ownerObjectPrefix(item), item.ownerUserId(), maxInputBytes); ParsedDocument document = parser.parse(stored.fileName(), item.mimeType(), stored.bytes()); List chunks = document.chunks(CHUNK_SIZE, CHUNK_OVERLAP); if (chunks.isEmpty()) { @@ -156,10 +157,12 @@ public class PersonalIngestionWorker { } private static StoredObjectReader defaultReader(ISysOssService ossService) { - return (ossId, maxBytes) -> { + return (ossId, expectedPrefix, ownerUserId, maxBytes) -> { SysOssVo object = ossService.getById(ossId); if (object == null || object.getFileName() == null || object.getFileName().isBlank() - || object.getService() == null || object.getService().isBlank()) { + || object.getService() == null || object.getService().isBlank() + || object.getCreateBy() == null || object.getCreateBy() != ownerUserId + || !object.getFileName().startsWith(expectedPrefix)) { throw new IOException("personal source object is unavailable"); } OssClient storage = OssFactory.instance(object.getService()); @@ -246,9 +249,13 @@ public class PersonalIngestionWorker { return safe.isBlank() ? "personal-object" : safe; } + private static String ownerObjectPrefix(Item item) { + return "personal/" + item.tenantId() + "/" + item.ownerUserId() + "/" + item.id() + "/"; + } + @FunctionalInterface public interface StoredObjectReader { - StoredObject read(long ossId, long maxBytes) throws Exception; + StoredObject read(long ossId, String expectedPrefix, long ownerUserId, long maxBytes) throws Exception; } public record StoredObject(String fileName, byte[] bytes) { diff --git a/backend/ruoyi-modules/ruoyi-aihr/src/test/java/org/dromara/aihr/personal/PersonalIngestionServiceTest.java b/backend/ruoyi-modules/ruoyi-aihr/src/test/java/org/dromara/aihr/personal/PersonalIngestionServiceTest.java index 0a06ec5d..b89de717 100644 --- a/backend/ruoyi-modules/ruoyi-aihr/src/test/java/org/dromara/aihr/personal/PersonalIngestionServiceTest.java +++ b/backend/ruoyi-modules/ruoyi-aihr/src/test/java/org/dromara/aihr/personal/PersonalIngestionServiceTest.java @@ -1,9 +1,11 @@ package org.dromara.aihr.personal; +import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import org.dromara.aihr.personal.domain.PersonalAssistantDto.ItemCreatedResponse; import org.dromara.aihr.personal.domain.PersonalAssistantDto.TextItemRequest; import org.dromara.aihr.personal.service.PersonalIngestionService; +import org.dromara.aihr.personal.service.PersonalIngestionService.PersonalObjectStore; import org.dromara.aihr.personal.service.PersonalSpaceService; import org.dromara.aihr.personal.support.PersonalKnowledgeProperties; import org.dromara.aihr.personal.support.PersonalOwner; @@ -16,10 +18,7 @@ import org.mockito.ArgumentCaptor; import org.mockito.InOrder; import org.springframework.aop.framework.ProxyFactory; import org.springframework.jdbc.core.JdbcTemplate; -import org.springframework.jdbc.core.PreparedStatementCreator; -import org.springframework.jdbc.support.KeyHolder; import org.springframework.mock.web.MockMultipartFile; -import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.TransactionDefinition; import org.springframework.transaction.annotation.AnnotationTransactionAttributeSource; import org.springframework.transaction.interceptor.TransactionInterceptor; @@ -32,17 +31,20 @@ import java.time.LocalDateTime; import java.util.List; import java.util.Map; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.contains; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.inOrder; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.verifyNoInteractions; import static org.mockito.Mockito.when; @@ -54,224 +56,240 @@ class PersonalIngestionServiceTest { @Test void failedItemRetriesThroughOwnerScopedQueuedState() { - JdbcTemplate jdbc = mock(JdbcTemplate.class); - when(jdbc.update(contains("status = 'QUEUED'"), eq("000000"), eq(101L), eq(9L))) + Fixture fixture = fixture(100L); + when(fixture.jdbc.update(contains("status = 'QUEUED'"), eq("000000"), eq(101L), eq(9L))) .thenReturn(1); - PersonalIngestionService service = service(jdbc, mock(PersonalSpaceService.class), mock(ISysOssService.class)); - service.retry(OWNER, 9L); + fixture.service.retry(OWNER, 9L); - verify(jdbc).update(contains("status = 'QUEUED'"), eq("000000"), eq(101L), eq(9L)); + verify(fixture.jdbc).update(contains("status = 'QUEUED'"), eq("000000"), eq(101L), eq(9L)); } @Test void retryDoesNotDiscloseMissingOrForeignItem() { - JdbcTemplate jdbc = mock(JdbcTemplate.class); - PersonalIngestionService service = service(jdbc, mock(PersonalSpaceService.class), mock(ISysOssService.class)); + Fixture fixture = fixture(100L); - ServiceException error = assertThrows(ServiceException.class, () -> service.retry(OWNER, 9L)); + ServiceException error = assertThrows(ServiceException.class, () -> fixture.service.retry(OWNER, 9L)); assertEquals("PERSONAL_ITEM_NOT_FOUND", error.getMessage()); } @Test - void fileLargerThanConfiguredLimitIsRejectedBeforeOssUpload() { - JdbcTemplate jdbc = mock(JdbcTemplate.class); - ISysOssService oss = mock(ISysOssService.class); - MockMultipartFile file = new MockMultipartFile( + void invalidFilesAreRejectedBeforeObjectStorage() { + Fixture fixture = fixture(100L); + MockMultipartFile large = new MockMultipartFile( "file", "large.pdf", "application/pdf", new byte[21 * 1024 * 1024]); - PersonalIngestionService service = service(jdbc, mock(PersonalSpaceService.class), oss); - ServiceException error = assertThrows(ServiceException.class, - () -> service.createFile(OWNER, file, null, null)); - - assertEquals("PERSONAL_FILE_TOO_LARGE", error.getMessage()); - verifyNoInteractions(oss); - verifyNoInteractions(jdbc); - } - - @Test - void unsupportedAndEmptyFilesAreRejectedBeforeOssUpload() { - JdbcTemplate jdbc = mock(JdbcTemplate.class); - ISysOssService oss = mock(ISysOssService.class); - PersonalIngestionService service = service(jdbc, mock(PersonalSpaceService.class), oss); - - assertThrows(ServiceException.class, () -> service.createFile( - OWNER, new MockMultipartFile("file", "script.exe", "application/octet-stream", new byte[]{1}), null, null)); - assertThrows(ServiceException.class, () -> service.createFile( + assertEquals("PERSONAL_FILE_TOO_LARGE", assertThrows(ServiceException.class, + () -> fixture.service.createFile(OWNER, large, null, null)).getMessage()); + assertThrows(ServiceException.class, () -> fixture.service.createFile( + OWNER, new MockMultipartFile("file", "script.exe", "application/octet-stream", new byte[]{1}), + null, null)); + assertThrows(ServiceException.class, () -> fixture.service.createFile( OWNER, new MockMultipartFile("file", "empty.txt", "text/plain", new byte[0]), null, null)); - verifyNoInteractions(oss); - verifyNoInteractions(jdbc); + verifyNoInteractions(fixture.store, fixture.oss, fixture.jdbc, fixture.spaces); } @Test - void createTextReservesUploadsInsertsAndMutatesCountersOnce() { - JdbcTemplate jdbc = mock(JdbcTemplate.class); - PersonalSpaceService spaces = mock(PersonalSpaceService.class); - ISysOssService oss = mock(ISysOssService.class); - when(spaces.reserve(eq(OWNER), eq(18L))).thenReturn(7L); - when(jdbc.queryForList(contains("content_hash"), eq("000000"), eq(101L), any(String.class))) - .thenReturn(List.of()); - when(oss.upload(any(java.io.File.class))).thenReturn(oss(81L)); - generatedId(jdbc, 99L); - when(jdbc.update(contains("used_bytes = used_bytes +"), eq(18L), eq("000000"), eq(101L), eq(7L))) - .thenReturn(1); - PersonalIngestionService service = service(jdbc, spaces, oss); + void textCreationLocksBeforeDedupeAndUsesIsolatedObjectKeyAndExtBinding() throws Exception { + Fixture fixture = fixture(100L); + stubSuccessfulCreate(fixture, OWNER, 7L, 18L, 81L); - ItemCreatedResponse response = service.createText( - OWNER, new TextItemRequest("周报", "保洁巡检记录", LocalDateTime.of(2026, 7, 12, 9, 0), List.of("保洁"))); - - assertEquals(99L, response.itemId()); - assertEquals("QUEUED", response.status()); - assertEquals(null, response.duplicateOf()); - InOrder order = inOrder(spaces, oss, jdbc); - order.verify(spaces).reserve(OWNER, 18L); - order.verify(oss).upload(any(java.io.File.class)); - order.verify(jdbc).update(any(PreparedStatementCreator.class), any(KeyHolder.class)); - order.verify(jdbc).update(contains("used_bytes = used_bytes +"), eq(18L), eq("000000"), eq(101L), eq(7L)); - assertNotNull(PersonalIngestionService.class.getAnnotation(org.springframework.stereotype.Service.class)); - assertNotNull(method("createText", PersonalOwner.class, TextItemRequest.class).getAnnotation(Transactional.class)); - } - - @Test - void createFileDoesNotDeduplicateAcrossOwners() throws Exception { - JdbcTemplate jdbc = mock(JdbcTemplate.class); - PersonalSpaceService spaces = mock(PersonalSpaceService.class); - ISysOssService oss = mock(ISysOssService.class); - MockMultipartFile file = new MockMultipartFile( - "file", "notes.txt", "text/plain", "same".getBytes(StandardCharsets.UTF_8)); - when(jdbc.queryForList(contains("content_hash"), eq("000000"), eq(101L), any(String.class))) - .thenReturn(List.of()); - when(spaces.reserve(OWNER, 4L)).thenReturn(7L); - when(oss.upload(any(org.springframework.web.multipart.MultipartFile.class))).thenReturn(oss(82L)); - generatedId(jdbc, 100L); - when(jdbc.update(contains("used_bytes = used_bytes +"), eq(4L), eq("000000"), eq(101L), eq(7L))) - .thenReturn(1); - PersonalIngestionService service = service(jdbc, spaces, oss); - - ItemCreatedResponse response = service.createFile(OWNER, file, null, null); + ItemCreatedResponse response = fixture.service.createText(OWNER, + new TextItemRequest("周报", "保洁巡检记录", LocalDateTime.of(2026, 7, 12, 9, 0), List.of("保洁"))); assertEquals(100L, response.itemId()); - ArgumentCaptor uploaded = - ArgumentCaptor.forClass(org.springframework.web.multipart.MultipartFile.class); - verify(oss).upload(uploaded.capture()); - assertTrue(uploaded.getValue().getOriginalFilename().matches("personal-[0-9a-f-]+\\.txt")); - assertEquals("same", new String(uploaded.getValue().getBytes(), StandardCharsets.UTF_8)); - verify(jdbc).queryForList(contains("content_hash"), eq("000000"), eq(101L), any(String.class)); - verify(jdbc, never()).queryForList(contains("content_hash"), eq("000000"), eq(202L), any(String.class)); + assertEquals("QUEUED", response.status()); + InOrder lockOrder = inOrder(fixture.spaces, fixture.jdbc); + lockOrder.verify(fixture.spaces).reserve(OWNER, 18L); + lockOrder.verify(fixture.jdbc).queryForList(contains("content_hash"), + eq("000000"), eq(101L), eq(7L), anyString()); + + ArgumentCaptor key = ArgumentCaptor.forClass(String.class); + verify(fixture.store).upload(eq(OWNER), eq(100L), key.capture(), eq("txt"), eq("text/plain"), + any(byte[].class)); + assertTrue(key.getValue().matches("personal/000000/101/100/[0-9a-f]{32}\\.txt")); + + ArgumentCaptor ext = ArgumentCaptor.forClass(String.class); + verify(fixture.jdbc).update(contains("update sys_oss"), ext.capture(), eq(101L), eq("000000"), + eq(81L), eq(101L), eq(key.getValue())); + JsonNode extJson = new ObjectMapper().readTree(ext.getValue()); + assertEquals("personal", extJson.path("source").asText()); + assertEquals(100L, extJson.path("itemId").asLong()); + verify(fixture.jdbc).update(contains("insert into aihr_personal_item"), eq(100L), eq("000000"), + eq(7L), eq(101L), eq("TEXT"), eq("周报"), eq(81L), eq("text/plain"), eq(18L), + anyString(), anyString(), any(LocalDateTime.class)); + verify(fixture.jdbc).update(contains("used_bytes = used_bytes +"), + eq(18L), eq("000000"), eq(101L), eq(7L)); } @Test - void duplicateForSameOwnerReturnsExistingItemWithoutReservationOrUpload() { - JdbcTemplate jdbc = mock(JdbcTemplate.class); - PersonalSpaceService spaces = mock(PersonalSpaceService.class); - ISysOssService oss = mock(ISysOssService.class); + void sameOwnerDuplicateIsCheckedUnderLockAndDoesNotUploadOrIncrementCounters() { + Fixture fixture = fixture(100L); + when(fixture.spaces.reserve(OWNER, 4L)).thenReturn(7L); + when(fixture.jdbc.queryForList(contains("content_hash"), + eq("000000"), eq(101L), eq(7L), anyString())) + .thenReturn(List.of(Map.of("id", 77L, "status", "READY"))); MockMultipartFile file = new MockMultipartFile( "file", "notes.txt", "text/plain", "same".getBytes(StandardCharsets.UTF_8)); - when(jdbc.queryForList(contains("content_hash"), eq("000000"), eq(101L), any(String.class))) - .thenReturn(List.of(Map.of("id", 77L, "status", "READY"))); - PersonalIngestionService service = service(jdbc, spaces, oss); - ItemCreatedResponse response = service.createFile(OWNER, file, null, null); + ItemCreatedResponse response = fixture.service.createFile(OWNER, file, null, null); assertEquals(77L, response.itemId()); assertEquals(77L, response.duplicateOf()); - assertEquals("READY", response.status()); - verifyNoInteractions(spaces, oss); + InOrder order = inOrder(fixture.spaces, fixture.jdbc); + order.verify(fixture.spaces).reserve(OWNER, 4L); + order.verify(fixture.jdbc).queryForList(contains("content_hash"), + eq("000000"), eq(101L), eq(7L), anyString()); + verifyNoInteractions(fixture.store, fixture.oss); + verify(fixture.jdbc, never()).update(contains("used_bytes = used_bytes +"), + any(), any(), any(), any()); } @Test - void failedPersistenceRegistersBestEffortOssCleanup() { - JdbcTemplate jdbc = mock(JdbcTemplate.class); - PersonalSpaceService spaces = mock(PersonalSpaceService.class); - ISysOssService oss = mock(ISysOssService.class); + void sameContentAcrossOwnersCreatesSeparateObjects() { + AtomicLong ids = new AtomicLong(100L); + Fixture fixture = fixture(ids::getAndIncrement); + PersonalOwner other = new PersonalOwner("000000", 202L, "ext-202"); + when(fixture.spaces.reserve(OWNER, 4L)).thenReturn(7L); + when(fixture.spaces.reserve(other, 4L)).thenReturn(8L); + when(fixture.jdbc.queryForList(contains("content_hash"), any(), any(), any(), any())) + .thenReturn(List.of()); + when(fixture.store.upload(any(), anyLong(), anyString(), eq("txt"), eq("text/plain"), + any(byte[].class))).thenAnswer(invocation -> oss(80L + invocation.getArgument(1), + invocation.getArgument(2))); + stubPersistence(fixture.jdbc); MockMultipartFile file = new MockMultipartFile( "file", "notes.txt", "text/plain", "same".getBytes(StandardCharsets.UTF_8)); - when(jdbc.queryForList(contains("content_hash"), eq("000000"), eq(101L), any(String.class))) - .thenReturn(List.of()); - when(spaces.reserve(OWNER, 4L)).thenReturn(7L); - SysOssVo rolledBack = oss(82L); - rolledBack.setService(null); - when(oss.upload(any(org.springframework.web.multipart.MultipartFile.class))).thenReturn(rolledBack); - when(jdbc.update(any(PreparedStatementCreator.class), any(KeyHolder.class))) + + ItemCreatedResponse first = fixture.service.createFile(OWNER, file, null, null); + ItemCreatedResponse second = fixture.service.createFile(other, file, null, null); + + assertEquals(100L, first.itemId()); + assertEquals(101L, second.itemId()); + verify(fixture.jdbc).queryForList(contains("content_hash"), + eq("000000"), eq(101L), eq(7L), anyString()); + verify(fixture.jdbc).queryForList(contains("content_hash"), + eq("000000"), eq(202L), eq(8L), anyString()); + verify(fixture.store, times(2)).upload(any(), anyLong(), anyString(), eq("txt"), eq("text/plain"), + any(byte[].class)); + } + + @Test + void failedItemPersistenceRollsBackAndCleansPhysicalObject() { + Fixture fixture = fixture(100L); + when(fixture.spaces.reserve(OWNER, 4L)).thenReturn(7L); + when(fixture.jdbc.queryForList(contains("content_hash"), + eq("000000"), eq(101L), eq(7L), anyString())).thenReturn(List.of()); + SysOssVo uploaded = oss(81L, "personal/000000/101/100/a.txt"); + when(fixture.store.upload(any(), anyLong(), anyString(), anyString(), anyString(), any(byte[].class))) + .thenReturn(uploaded); + when(fixture.jdbc.update(contains("update sys_oss"), any(), any(), any(), any(), any(), any())) + .thenReturn(1); + when(fixture.jdbc.update(contains("insert into aihr_personal_item"), + any(), any(), any(), any(), any(), any(), any(), any(), any(), any(), any(), any())) .thenThrow(new IllegalStateException("db failed")); TestTransactionManager transactions = new TestTransactionManager(); - PersonalIngestionService service = transactionalProxy(service(jdbc, spaces, oss), transactions); + PersonalIngestionService proxy = transactionalProxy(fixture.service, transactions); + MockMultipartFile file = new MockMultipartFile( + "file", "notes.txt", "text/plain", "same".getBytes(StandardCharsets.UTF_8)); - assertThrows(IllegalStateException.class, () -> service.createFile(OWNER, file, null, null)); + assertThrows(IllegalStateException.class, () -> proxy.createFile(OWNER, file, null, null)); - verify(oss).deleteWithValidByIds(eq(List.of(82L)), eq(false)); assertEquals(1, transactions.rollbacks); + verify(fixture.store).deletePhysical(uploaded); + verify(fixture.oss).deleteWithValidByIds(List.of(81L), false); } @Test - void reservationAndCounterMutationRunInsideOneOuterTransaction() { - JdbcTemplate jdbc = mock(JdbcTemplate.class); - PersonalSpaceService spaces = mock(PersonalSpaceService.class); - ISysOssService oss = mock(ISysOssService.class); - MockMultipartFile file = new MockMultipartFile( - "file", "notes.txt", "text/plain", "same".getBytes(StandardCharsets.UTF_8)); + void reservationAndCountersShareOneOuterTransaction() { + Fixture fixture = fixture(100L); AtomicBoolean reserveInTransaction = new AtomicBoolean(); AtomicBoolean counterInTransaction = new AtomicBoolean(); - when(jdbc.queryForList(contains("content_hash"), eq("000000"), eq(101L), any(String.class))) - .thenReturn(List.of()); - when(spaces.reserve(OWNER, 4L)).thenAnswer(invocation -> { + when(fixture.spaces.reserve(OWNER, 4L)).thenAnswer(invocation -> { reserveInTransaction.set(TransactionSynchronizationManager.isActualTransactionActive()); return 7L; }); - when(oss.upload(any(org.springframework.web.multipart.MultipartFile.class))).thenReturn(oss(82L)); - generatedId(jdbc, 100L); - when(jdbc.update(contains("used_bytes = used_bytes +"), eq(4L), eq("000000"), eq(101L), eq(7L))) + when(fixture.jdbc.queryForList(contains("content_hash"), any(), any(), any(), any())) + .thenReturn(List.of()); + when(fixture.store.upload(any(), anyLong(), anyString(), anyString(), anyString(), any(byte[].class))) + .thenReturn(oss(81L, "personal/000000/101/100/a.txt")); + when(fixture.jdbc.update(contains("update sys_oss"), any(), any(), any(), any(), any(), any())) + .thenReturn(1); + when(fixture.jdbc.update(contains("insert into aihr_personal_item"), + any(), any(), any(), any(), any(), any(), any(), any(), any(), any(), any(), any())) + .thenReturn(1); + when(fixture.jdbc.update(contains("used_bytes = used_bytes +"), any(), any(), any(), any())) .thenAnswer(invocation -> { counterInTransaction.set(TransactionSynchronizationManager.isActualTransactionActive()); return 1; }); TestTransactionManager transactions = new TestTransactionManager(); - PersonalIngestionService service = transactionalProxy(service(jdbc, spaces, oss), transactions); + PersonalIngestionService proxy = transactionalProxy(fixture.service, transactions); + MockMultipartFile file = new MockMultipartFile( + "file", "notes.txt", "text/plain", "same".getBytes(StandardCharsets.UTF_8)); - service.createFile(OWNER, file, null, null); + proxy.createFile(OWNER, file, null, null); assertTrue(reserveInTransaction.get()); assertTrue(counterInTransaction.get()); assertEquals(1, transactions.commits); - assertEquals(0, transactions.rollbacks); } - private static PersonalIngestionService service(JdbcTemplate jdbc, PersonalSpaceService spaces, ISysOssService oss) { - return new PersonalIngestionService(jdbc, spaces, properties(), oss, new ObjectMapper()); + @Test + void unsafeTenantIsRejectedBeforeStorage() { + Fixture fixture = fixture(100L); + PersonalOwner unsafe = new PersonalOwner("../000000", 101L, null); + + assertEquals("PERSONAL_OWNER_INVALID", assertThrows(ServiceException.class, + () -> fixture.service.createText(unsafe, new TextItemRequest("x", "body", null, List.of()))) + .getMessage()); + + verifyNoInteractions(fixture.spaces, fixture.store, fixture.jdbc, fixture.oss); } - private static PersonalKnowledgeProperties properties() { - return new PersonalKnowledgeProperties(); + private static Fixture fixture(long itemId) { + return fixture(() -> itemId); } - private static SysOssVo oss(long id) { + private static Fixture fixture(java.util.function.LongSupplier itemIds) { + JdbcTemplate jdbc = mock(JdbcTemplate.class); + PersonalSpaceService spaces = mock(PersonalSpaceService.class); + ISysOssService ossService = mock(ISysOssService.class); + PersonalObjectStore store = mock(PersonalObjectStore.class); + PersonalIngestionService service = PersonalIngestionService.forTest( + jdbc, spaces, new PersonalKnowledgeProperties(), ossService, new ObjectMapper(), store, itemIds); + return new Fixture(jdbc, spaces, ossService, store, service); + } + + private static void stubSuccessfulCreate(Fixture fixture, PersonalOwner owner, long spaceId, long bytes, + long ossId) { + when(fixture.spaces.reserve(owner, bytes)).thenReturn(spaceId); + when(fixture.jdbc.queryForList(contains("content_hash"), + eq(owner.tenantId()), eq(owner.userId()), eq(spaceId), anyString())).thenReturn(List.of()); + when(fixture.store.upload(eq(owner), anyLong(), anyString(), anyString(), anyString(), any(byte[].class))) + .thenAnswer(invocation -> oss(ossId, invocation.getArgument(2))); + stubPersistence(fixture.jdbc); + } + + private static void stubPersistence(JdbcTemplate jdbc) { + when(jdbc.update(contains("update sys_oss"), any(), any(), any(), any(), any(), any())).thenReturn(1); + when(jdbc.update(contains("insert into aihr_personal_item"), + any(), any(), any(), any(), any(), any(), any(), any(), any(), any(), any(), any())).thenReturn(1); + when(jdbc.update(contains("used_bytes = used_bytes +"), any(), any(), any(), any())).thenReturn(1); + } + + private static SysOssVo oss(long id, String fileName) { SysOssVo result = new SysOssVo(); result.setOssId(id); - result.setFileName("personal/random.txt"); - result.setOriginalName("random.txt"); + result.setFileName(fileName); + result.setOriginalName(fileName.substring(fileName.lastIndexOf('/') + 1)); result.setService("minio"); - result.setUrl("https://private.invalid/random.txt"); + result.setUrl("https://private.invalid/" + fileName); return result; } - private static void generatedId(JdbcTemplate jdbc, long id) { - when(jdbc.update(any(PreparedStatementCreator.class), any(KeyHolder.class))).thenAnswer(invocation -> { - KeyHolder holder = invocation.getArgument(1); - holder.getKeyList().add(Map.of("GENERATED_KEY", id)); - return 1; - }); - } - - private static java.lang.reflect.Method method(String name, Class... types) { - try { - return PersonalIngestionService.class.getMethod(name, types); - } catch (NoSuchMethodException e) { - throw new AssertionError(e); - } - } - private static PersonalIngestionService transactionalProxy(PersonalIngestionService target, TestTransactionManager transactionManager) { ProxyFactory factory = new ProxyFactory(target); @@ -284,6 +302,10 @@ class PersonalIngestionServiceTest { return (PersonalIngestionService) factory.getProxy(); } + private record Fixture(JdbcTemplate jdbc, PersonalSpaceService spaces, ISysOssService oss, + PersonalObjectStore store, PersonalIngestionService service) { + } + private static final class TestTransactionManager extends AbstractPlatformTransactionManager { private int commits; private int rollbacks; diff --git a/backend/ruoyi-modules/ruoyi-aihr/src/test/java/org/dromara/aihr/personal/PersonalIngestionWorkerTest.java b/backend/ruoyi-modules/ruoyi-aihr/src/test/java/org/dromara/aihr/personal/PersonalIngestionWorkerTest.java index c844d4a8..fc106737 100644 --- a/backend/ruoyi-modules/ruoyi-aihr/src/test/java/org/dromara/aihr/personal/PersonalIngestionWorkerTest.java +++ b/backend/ruoyi-modules/ruoyi-aihr/src/test/java/org/dromara/aihr/personal/PersonalIngestionWorkerTest.java @@ -44,8 +44,12 @@ class PersonalIngestionWorkerTest { .thenReturn(1); PersonalIngestionWorker worker = PersonalIngestionWorker.forTest( jdbc, mock(ISysOssService.class), parser, transactions, - (ossId, maxBytes) -> new PersonalIngestionWorker.StoredObject( - "notes.txt", "same".getBytes(StandardCharsets.UTF_8))); + (ossId, prefix, ownerUserId, maxBytes) -> { + assertTrue(prefix.equals("personal/000000/101/9/")); + assertTrue(ownerUserId == 101L); + return new PersonalIngestionWorker.StoredObject( + "notes.txt", "same".getBytes(StandardCharsets.UTF_8)); + }); assertTrue(worker.processNext()); @@ -61,7 +65,8 @@ class PersonalIngestionWorkerTest { when(jdbc.queryForList(contains("status = 'QUEUED'"))).thenReturn(List.of(item())); PersonalIngestionWorker worker = PersonalIngestionWorker.forTest( jdbc, mock(ISysOssService.class), mock(KnowledgeDocumentParser.class), immediateTransactions(), - (ossId, maxBytes) -> new PersonalIngestionWorker.StoredObject("notes.txt", new byte[]{1})); + (ossId, prefix, ownerUserId, maxBytes) -> + new PersonalIngestionWorker.StoredObject("notes.txt", new byte[]{1})); assertFalse(worker.processNext()); @@ -80,7 +85,8 @@ class PersonalIngestionWorkerTest { KnowledgeDocumentParser.Failure.INVALID, "secret parser detail")); PersonalIngestionWorker worker = PersonalIngestionWorker.forTest( jdbc, mock(ISysOssService.class), parser, immediateTransactions(), - (ossId, maxBytes) -> new PersonalIngestionWorker.StoredObject("notes.txt", new byte[]{1})); + (ossId, prefix, ownerUserId, maxBytes) -> + new PersonalIngestionWorker.StoredObject("notes.txt", new byte[]{1})); assertTrue(worker.processNext());