猪群管理聚合服务 增加调栏管理

internal/app/service/pig_batch_service.go

@@ -65,31 +65,31 @@ func (s *pigBatchService) CreatePigBatch(dto *dto.PigBatchCreateDTO) (*dto.PigBa
 	createdBatch, err := s.domainService.CreatePigBatch(batch)
 	if err != nil {
 		s.logger.Errorf("应用层: 创建猪批次失败: %v", err)
-		return nil, err // 将领域层的错误传递上去
+		return nil, mapDomainError(err)
 	}
 
 	// 3. 领域模型 -> DTO
 	return s.toPigBatchResponseDTO(createdBatch), nil
 }
 
-// GetPigBatch 从领域服务获取数据并转换为DTO。
+// GetPigBatch 从领域服务获取数据并转换为DTO，同时处理错误转换。
 func (s *pigBatchService) GetPigBatch(id uint) (*dto.PigBatchResponseDTO, error) {
 	batch, err := s.domainService.GetPigBatch(id)
 	if err != nil {
 		s.logger.Warnf("应用层: 获取猪批次失败, ID: %d, 错误: %v", id, err)
-		return nil, err
+		return nil, mapDomainError(err)
 	}
 
 	return s.toPigBatchResponseDTO(batch), nil
 }
 
-// UpdatePigBatch 协调获取、更新和保存的流程。
+// UpdatePigBatch 协调获取、更新和保存的流程，并处理错误转换。
 func (s *pigBatchService) UpdatePigBatch(id uint, dto *dto.PigBatchUpdateDTO) (*dto.PigBatchResponseDTO, error) {
 	// 1. 先获取最新的领域模型
 	existingBatch, err := s.domainService.GetPigBatch(id)
 	if err != nil {
 		s.logger.Warnf("应用层: 更新猪批次失败，获取原批次信息错误, ID: %d, 错误: %v", id, err)
-		return nil, err
+		return nil, mapDomainError(err)
 	}
 
 	// 2. 将DTO中的变更应用到模型上
@@ -116,19 +116,19 @@ func (s *pigBatchService) UpdatePigBatch(id uint, dto *dto.PigBatchUpdateDTO) (*
 	updatedBatch, err := s.domainService.UpdatePigBatch(existingBatch)
 	if err != nil {
 		s.logger.Errorf("应用层: 更新猪批次失败, ID: %d, 错误: %v", id, err)
-		return nil, err
+		return nil, mapDomainError(err)
 	}
 
 	// 4. 转换并返回结果
 	return s.toPigBatchResponseDTO(updatedBatch), nil
 }
 
-// DeletePigBatch 将删除操作委托给领域服务。
+// DeletePigBatch 将删除操作委托给领域服务，并转换领域错误为应用层错误。
 func (s *pigBatchService) DeletePigBatch(id uint) error {
 	err := s.domainService.DeletePigBatch(id)
 	if err != nil {
 		s.logger.Errorf("应用层: 删除猪批次失败, ID: %d, 错误: %v", id, err)
-		return err
+		return mapDomainError(err)
 	}
 	return nil
 }
@@ -138,7 +138,7 @@ func (s *pigBatchService) ListPigBatches(isActive *bool) ([]*dto.PigBatchRespons
 	batches, err := s.domainService.ListPigBatches(isActive)
 	if err != nil {
 		s.logger.Errorf("应用层: 批量查询猪批次失败: %v", err)
-		return nil, err
+		return nil, mapDomainError(err)
 	}
 
 	var responseDTOs []*dto.PigBatchResponseDTO
@@ -149,12 +149,12 @@ func (s *pigBatchService) ListPigBatches(isActive *bool) ([]*dto.PigBatchRespons
 	return responseDTOs, nil
 }
 
-// UpdatePigBatchPens 将关联猪栏的复杂操作委托给领域服务。
+// UpdatePigBatchPens 将关联猪栏的复杂操作委托给领域服务，并处理错误转换。
 func (s *pigBatchService) UpdatePigBatchPens(batchID uint, desiredPenIDs []uint) error {
 	err := s.domainService.UpdatePigBatchPens(batchID, desiredPenIDs)
 	if err != nil {
 		s.logger.Errorf("应用层: 更新猪批次猪栏关联失败, 批次ID: %d, 错误: %v", batchID, err)
-		return err
+		return mapDomainError(err)
 	}
 	return nil
 }

internal/app/service/pig_farm_service.go

@@ -11,15 +11,6 @@ import (
 	"gorm.io/gorm"
 )
 
-var (
-	ErrHouseContainsPens                = errors.New("无法删除包含猪栏的猪舍")
-	ErrHouseNotFound                    = errors.New("指定的猪舍不存在")
-	ErrPenInUse                         = errors.New("猪栏正在被活跃批次使用，无法删除")
-	ErrPenNotFound                      = errors.New("指定的猪栏不存在")
-	ErrPenStatusInvalidForOccupiedPen   = errors.New("猪栏已被批次使用，无法设置为非使用中状态")
-	ErrPenStatusInvalidForUnoccupiedPen = errors.New("猪栏未被批次使用，无法设置为使用中状态")
-)
-
 // PigFarmService 提供了猪场资产管理的业务逻辑
 type PigFarmService interface {
 	// PigHouse methods

internal/app/service/pig_service.go

@@ -1 +1,49 @@
 package service
+
+import (
+	"errors"
+
+	domain_pig "git.huangwc.com/pig/pig-farm-controller/internal/domain/pig"
+)
+
+var (
+	ErrHouseContainsPens                = errors.New("无法删除包含猪栏的猪舍")
+	ErrHouseNotFound                    = errors.New("指定的猪舍不存在")
+	ErrPenInUse                         = errors.New("猪栏正在被活跃批次使用，无法删除")
+	ErrPenNotFound                      = errors.New("指定的猪栏不存在")
+	ErrPenStatusInvalidForOccupiedPen   = errors.New("猪栏已被批次使用，无法设置为非使用中状态")
+	ErrPenStatusInvalidForUnoccupiedPen = errors.New("猪栏未被批次使用，无法设置为使用中状态")
+	ErrPigBatchNotFound                 = errors.New("指定的猪批次不存在")
+	ErrPigBatchActive                   = errors.New("活跃的猪批次不能被删除")
+	ErrPigBatchNotActive                = errors.New("猪批次不处于活跃状态，无法修改关联猪栏")
+	ErrPenOccupiedByOtherBatch          = errors.New("猪栏已被其他批次使用")
+	ErrPenStatusInvalidForAllocation    = errors.New("猪栏状态不允许分配")
+	ErrPenNotAssociatedWithBatch        = errors.New("猪栏未与该批次关联")
+)
+
+// mapDomainError 将领域层的错误转换为应用服务层的公共错误。
+func mapDomainError(err error) error {
+	if err == nil {
+		return nil
+	}
+
+	switch {
+	case errors.Is(err, domain_pig.ErrPigBatchNotFound):
+		return ErrPigBatchNotFound
+	case errors.Is(err, domain_pig.ErrPigBatchActive):
+		return ErrPigBatchActive
+	case errors.Is(err, domain_pig.ErrPigBatchNotActive):
+		return ErrPigBatchNotActive
+	case errors.Is(err, domain_pig.ErrPenOccupiedByOtherBatch):
+		return ErrPenOccupiedByOtherBatch
+	case errors.Is(err, domain_pig.ErrPenStatusInvalidForAllocation):
+		return ErrPenStatusInvalidForAllocation
+	case errors.Is(err, domain_pig.ErrPenNotAssociatedWithBatch):
+		return ErrPenNotAssociatedWithBatch
+	case errors.Is(err, domain_pig.ErrPenNotFound):
+		return ErrPenNotFound
+	// 可以添加更多领域错误到应用层错误的映射
+	default:
+		return err // 对于未知的领域错误，直接返回
+	}
+}

internal/core/application.go

@@ -74,12 +74,14 @@ func NewApplication(configPath string) (*Application, error) {
 	pendingCollectionRepo := repository.NewGormPendingCollectionRepository(storage.GetDB())
 	userActionLogRepo := repository.NewGormUserActionLogRepository(storage.GetDB())
 	pigBatchRepo := repository.NewGormPigBatchRepository(storage.GetDB())
+	penRepo := repository.NewPenRepository(storage.GetDB())
 
 	// 初始化事务管理器
 	unitOfWork := repository.NewGormUnitOfWork(storage.GetDB(), logger)
 
-	// 初始化猪群管理服务
+	// 初始化猪群管理领域
-	pigBatchDomain := pig.NewPigBatchService(pigBatchRepo, pigFarmRepo, unitOfWork)
+	penTransferManager := pig.NewPenTransferManager(penRepo)
+	pigBatchDomain := pig.NewPigBatchService(pigBatchRepo, unitOfWork, penTransferManager)
 
 	// --- 业务逻辑处理器初始化 ---
 	pigFarmService := service.NewPigFarmService(pigFarmRepo, unitOfWork, logger)

internal/domain/pig/pen_transfer_manager.go

@@ -0,0 +1,63 @@
+package pig
+
+import (
+	"git.huangwc.com/pig/pig-farm-controller/internal/infra/models"
+	"git.huangwc.com/pig/pig-farm-controller/internal/infra/repository"
+	"gorm.io/gorm"
+)
+
+// PenTransferManager 定义了与猪只位置转移相关的底层数据库操作。
+// 它是一个内部服务，被主服务 PigBatchService 调用。
+type PenTransferManager interface {
+	// LogTransfer 在数据库中创建一条猪只迁移日志。
+	LogTransfer(tx *gorm.DB, log *models.PigTransferLog) error
+
+	// GetPenByID 用于获取猪栏的详细信息，供上层服务进行业务校验。
+	// 注意: 此方法依赖于您在 PenRepository 中添加对应的 GetPenByIDTx 方法。
+	GetPenByID(tx *gorm.DB, penID uint) (*models.Pen, error)
+
+	// GetPensByBatchID 获取一个猪群当前关联的所有猪栏。
+	// 注意: 此方法依赖于您在 PenRepository 中添加对应的 GetPensByBatchIDTx 方法。
+	GetPensByBatchID(tx *gorm.DB, batchID uint) ([]*models.Pen, error)
+
+	// UpdatePenFields 更新一个猪栏的指定字段。
+	// 注意: 此方法依赖于您在 PenRepository 中添加对应的 UpdatePenFieldsTx 方法。
+	UpdatePenFields(tx *gorm.DB, penID uint, updates map[string]interface{}) error
+}
+
+// penTransferManager 是 PenTransferManager 接口的具体实现。
+// 它作为调栏管理器，处理底层的数据库交互。
+type penTransferManager struct {
+	penRepo repository.PenRepository
+}
+
+// NewPenTransferManager 是 penTransferManager 的构造函数。
+func NewPenTransferManager(penRepo repository.PenRepository) PenTransferManager {
+	return &penTransferManager{
+		penRepo: penRepo,
+	}
+}
+
+// LogTransfer 实现了在数据库中创建迁移日志的逻辑。
+func (s *penTransferManager) LogTransfer(tx *gorm.DB, log *models.PigTransferLog) error {
+	// 直接使用事务对象创建记录。
+	return tx.Create(log).Error
+}
+
+// GetPenByID 实现了获取猪栏信息的逻辑。
+// 注意: 此处调用了一个假设存在的方法 GetPenByIDTx。
+func (s *penTransferManager) GetPenByID(tx *gorm.DB, penID uint) (*models.Pen, error) {
+	return s.penRepo.GetPenByIDTx(tx, penID)
+}
+
+// GetPensByBatchID 实现了获取猪群关联猪栏列表的逻辑。
+// 注意: 此处调用了一个假设存在的方法 GetPensByBatchIDTx。
+func (s *penTransferManager) GetPensByBatchID(tx *gorm.DB, batchID uint) ([]*models.Pen, error) {
+	return s.penRepo.GetPensByBatchIDTx(tx, batchID)
+}
+
+// UpdatePenFields 实现了更新猪栏字段的逻辑。
+// 注意: 此处调用了一个假设存在的方法 UpdatePenFieldsTx。
+func (s *penTransferManager) UpdatePenFields(tx *gorm.DB, penID uint, updates map[string]interface{}) error {
+	return s.penRepo.UpdatePenFieldsTx(tx, penID, updates)
+}

internal/domain/pig/pig_batch.go

@@ -21,6 +21,8 @@ var (
 	ErrPenStatusInvalidForAllocation = errors.New("猪栏状态不允许分配")
 	// ErrPenNotFound 表示猪栏不存在
 	ErrPenNotFound = errors.New("指定的猪栏不存在")
+	// ErrPenNotAssociatedWithBatch 表示猪栏未与该批次关联
+	ErrPenNotAssociatedWithBatch = errors.New("猪栏未与该批次关联")
 )
 
 // --- 领域服务接口 ---
@@ -28,6 +30,12 @@ var (
 // PigBatchService 定义了猪批次管理的核心业务逻辑接口。
 // 它抽象了所有与猪批次相关的操作，使得应用层可以依赖于此接口，而不是具体的实现。
 type PigBatchService interface {
+	// TransferPigsWithinBatch 处理同一个猪群内部的调栏业务。
+	TransferPigsWithinBatch(batchID uint, fromPenID uint, toPenID uint, quantity uint) error
+
+	// TransferPigsAcrossBatches 处理跨猪群的调栏业务。
+	TransferPigsAcrossBatches(sourceBatchID uint, destBatchID uint, fromPenID uint, toPenID uint, quantity uint) error
+
 	// CreatePigBatch 创建一个新的猪批次。
 	CreatePigBatch(batch *models.PigBatch) (*models.PigBatch, error)
 

internal/domain/pig/pig_batch_service.go

@@ -3,33 +3,35 @@ package pig
 import (
 	"errors"
 	"fmt"
+	"time"
 
 	"git.huangwc.com/pig/pig-farm-controller/internal/infra/models"
 	"git.huangwc.com/pig/pig-farm-controller/internal/infra/repository"
+	"github.com/google/uuid"
 	"gorm.io/gorm"
 )
 
 // --- 领域服务实现 ---
 
 // pigBatchService 是 PigBatchService 接口的具体实现。
-// 它封装了业务逻辑所需的所有依赖，如数据库仓库和工作单元。
+// 它作为猪群领域的主服务，封装了所有业务逻辑。
 type pigBatchService struct {
 	pigBatchRepo repository.PigBatchRepository // 猪批次仓库
-	pigFarmRepo  repository.PigFarmRepository  // 猪场资产仓库 (包含猪栏操作)
 	uow          repository.UnitOfWork         // 工作单元，用于管理事务
+	transferSvc  PenTransferManager            // 调栏子服务
 }
 
 // NewPigBatchService 是 pigBatchService 的构造函数。
 // 它通过依赖注入的方式，创建并返回一个 PigBatchService 接口的实例。
 func NewPigBatchService(
 	pigBatchRepo repository.PigBatchRepository,
-	pigFarmRepo repository.PigFarmRepository,
 	uow repository.UnitOfWork,
+	transferSvc PenTransferManager,
 ) PigBatchService {
 	return &pigBatchService{
 		pigBatchRepo: pigBatchRepo,
-		pigFarmRepo:  pigFarmRepo,
 		uow:          uow,
+		transferSvc:  transferSvc,
 	}
 }
 
@@ -95,11 +97,13 @@ func (s *pigBatchService) ListPigBatches(isActive *bool) ([]*models.PigBatch, er
 }
 
 // UpdatePigBatchPens 实现了在事务中更新猪批次关联猪栏的复杂逻辑。
+// 它通过调用底层的 PenTransferManager 来执行数据库操作，从而保持了职责的清晰。
 func (s *pigBatchService) UpdatePigBatchPens(batchID uint, desiredPenIDs []uint) error {
 	// 使用工作单元来确保操作的原子性
 	return s.uow.ExecuteInTransaction(func(tx *gorm.DB) error {
 		// 1. 验证猪批次是否存在且活跃
-		pigBatch, err := s.pigFarmRepo.GetPigBatchByIDTx(tx, batchID)
+		// 注意: 此处依赖一个假设存在的 pigBatchRepo.GetPigBatchByIDTx 方法
+		pigBatch, err := s.pigBatchRepo.GetPigBatchByIDTx(tx, batchID)
 		if err != nil {
 			if errors.Is(err, gorm.ErrRecordNotFound) {
 				return ErrPigBatchNotFound
@@ -111,8 +115,8 @@ func (s *pigBatchService) UpdatePigBatchPens(batchID uint, desiredPenIDs []uint)
 			return ErrPigBatchNotActive
 		}
 
-		// 2. 获取当前关联的猪栏
+		// 2. 获取当前关联的猪栏 (通过子服务)
-		currentPens, err := s.pigFarmRepo.GetPensByBatchID(tx, batchID)
+		currentPens, err := s.transferSvc.GetPensByBatchID(tx, batchID)
 		if err != nil {
 			return fmt.Errorf("获取当前关联猪栏失败: %w", err)
 		}
@@ -120,7 +124,7 @@ func (s *pigBatchService) UpdatePigBatchPens(batchID uint, desiredPenIDs []uint)
 		currentPenMap := make(map[uint]models.Pen)
 		currentPenIDsSet := make(map[uint]struct{})
 		for _, pen := range currentPens {
-			currentPenMap[pen.ID] = pen
+			currentPenMap[pen.ID] = *pen
 			currentPenIDsSet[pen.ID] = struct{}{}
 		}
 
@@ -155,14 +159,15 @@ func (s *pigBatchService) UpdatePigBatchPens(batchID uint, desiredPenIDs []uint)
 				updates["status"] = models.PenStatusEmpty
 			}
 
-			if err := s.pigFarmRepo.UpdatePenFields(tx, penID, updates); err != nil {
+			if err := s.transferSvc.UpdatePenFields(tx, penID, updates); err != nil {
 				return fmt.Errorf("移除猪栏 %d 失败: %w", penID, err)
 			}
 		}
 
 		// 6. 处理添加猪栏的逻辑
 		for _, penID := range pensToAdd {
-			actualPen, err := s.pigFarmRepo.GetPenByIDTx(tx, penID)
+			// 通过子服务获取猪栏信息
+			actualPen, err := s.transferSvc.GetPenByID(tx, penID)
 			if err != nil {
 				if errors.Is(err, gorm.ErrRecordNotFound) {
 					return fmt.Errorf("猪栏 %d 不存在: %w", penID, ErrPenNotFound)
@@ -182,7 +187,7 @@ func (s *pigBatchService) UpdatePigBatchPens(batchID uint, desiredPenIDs []uint)
 				"pig_batch_id": &batchID,
 				"status":       models.PenStatusOccupied,
 			}
-			if err := s.pigFarmRepo.UpdatePenFields(tx, penID, updates); err != nil {
+			if err := s.transferSvc.UpdatePenFields(tx, penID, updates); err != nil {
 				return fmt.Errorf("添加猪栏 %d 失败: %w", penID, err)
 			}
 		}
@@ -190,3 +195,128 @@ func (s *pigBatchService) UpdatePigBatchPens(batchID uint, desiredPenIDs []uint)
 		return nil
 	})
 }
+
+// --- 新增的调栏业务实现 ---
+
+// executeTransferAndLog 是一个私有辅助方法，用于封装创建和记录迁移日志的通用逻辑。
+func (s *pigBatchService) executeTransferAndLog(tx *gorm.DB, fromBatchID, toBatchID, fromPenID, toPenID uint, quantity int, transferType string) error {
+	// 1. 生成关联ID
+	correlationID := uuid.New().String()
+
+	// 2. 创建调出日志
+	logOut := &models.PigTransferLog{
+		TransferTime:  time.Now(),
+		PigBatchID:    fromBatchID,
+		PenID:         fromPenID,
+		Quantity:      -quantity, // 调出为负数
+		Type:          transferType,
+		CorrelationID: correlationID,
+	}
+
+	// 3. 创建调入日志
+	logIn := &models.PigTransferLog{
+		TransferTime:  time.Now(),
+		PigBatchID:    toBatchID,
+		PenID:         toPenID,
+		Quantity:      quantity, // 调入为正数
+		Type:          transferType,
+		CorrelationID: correlationID,
+	}
+
+	// 4. 调用子服务记录日志
+	if err := s.transferSvc.LogTransfer(tx, logOut); err != nil {
+		return fmt.Errorf("记录调出日志失败: %w", err)
+	}
+	if err := s.transferSvc.LogTransfer(tx, logIn); err != nil {
+		return fmt.Errorf("记录调入日志失败: %w", err)
+	}
+
+	return nil
+}
+
+// TransferPigsWithinBatch 实现了同一个猪群内部的调栏业务。
+func (s *pigBatchService) TransferPigsWithinBatch(batchID uint, fromPenID uint, toPenID uint, quantity uint) error {
+	if fromPenID == toPenID {
+		return errors.New("源猪栏和目标猪栏不能相同")
+	}
+	if quantity == 0 {
+		return errors.New("迁移数量不能为零")
+	}
+
+	return s.uow.ExecuteInTransaction(func(tx *gorm.DB) error {
+		// 1. 核心业务规则校验
+		fromPen, err := s.transferSvc.GetPenByID(tx, fromPenID)
+		if err != nil {
+			return fmt.Errorf("获取源猪栏信息失败: %w", err)
+		}
+		toPen, err := s.transferSvc.GetPenByID(tx, toPenID)
+		if err != nil {
+			return fmt.Errorf("获取目标猪栏信息失败: %w", err)
+		}
+
+		if fromPen.PigBatchID == nil || *fromPen.PigBatchID != batchID {
+			return fmt.Errorf("源猪栏 %d 不属于指定的猪群 %d", fromPenID, batchID)
+		}
+		if toPen.PigBatchID != nil && *toPen.PigBatchID != batchID {
+			return fmt.Errorf("目标猪栏 %d 已被其他猪群占用", toPenID)
+		}
+
+		// 2. 调用通用辅助方法执行日志记录
+		err = s.executeTransferAndLog(tx, batchID, batchID, fromPenID, toPenID, int(quantity), "群内调栏")
+		if err != nil {
+			return err
+		}
+
+		// 3. 群内调栏，猪群总数不变
+		return nil
+	})
+}
+
+// TransferPigsAcrossBatches 实现了跨猪群的调栏业务。
+func (s *pigBatchService) TransferPigsAcrossBatches(sourceBatchID uint, destBatchID uint, fromPenID uint, toPenID uint, quantity uint) error {
+	if sourceBatchID == destBatchID {
+		return errors.New("源猪群和目标猪群不能相同")
+	}
+	if quantity == 0 {
+		return errors.New("迁移数量不能为零")
+	}
+
+	return s.uow.ExecuteInTransaction(func(tx *gorm.DB) error {
+		// 1. 核心业务规则校验
+		sourceBatch, err := s.pigBatchRepo.GetPigBatchByID(sourceBatchID)
+		if err != nil {
+			return fmt.Errorf("获取源猪群信息失败: %w", err)
+		}
+		destBatch, err := s.pigBatchRepo.GetPigBatchByID(destBatchID)
+		if err != nil {
+			return fmt.Errorf("获取目标猪群信息失败: %w", err)
+		}
+
+		fromPen, err := s.transferSvc.GetPenByID(tx, fromPenID)
+		if err != nil {
+			return fmt.Errorf("获取源猪栏信息失败: %w", err)
+		}
+		if fromPen.PigBatchID == nil || *fromPen.PigBatchID != sourceBatchID {
+			return fmt.Errorf("源猪栏 %d 不属于源猪群 %d", fromPenID, sourceBatchID)
+		}
+
+		// 2. 调用通用辅助方法执行日志记录
+		err = s.executeTransferAndLog(tx, sourceBatchID, destBatchID, fromPenID, toPenID, int(quantity), "跨群调栏")
+		if err != nil {
+			return err
+		}
+
+		// 3. 修改本聚合的数据（猪群总数）
+		sourceBatch.InitialCount -= int(quantity)
+		destBatch.InitialCount += int(quantity)
+
+		if _, _, err := s.pigBatchRepo.UpdatePigBatch(sourceBatch); err != nil {
+			return fmt.Errorf("更新源猪群数量失败: %w", err)
+		}
+		if _, _, err := s.pigBatchRepo.UpdatePigBatch(destBatch); err != nil {
+			return fmt.Errorf("更新目标猪群数量失败: %w", err)
+		}
+
+		return nil
+	})
+}

internal/infra/models/pig_transfer.go

@@ -0,0 +1,19 @@
+package models
+
+import (
+	"time"
+
+	"gorm.io/gorm"
+)
+
+// PigTransferLog 记录了每一次猪只数量在猪栏间的变动事件。
+// 它作为事件溯源的基础，用于推算任意时间点猪栏的猪只数量。
+type PigTransferLog struct {
+	gorm.Model
+	TransferTime  time.Time `json:"transfer_time"`  // 迁移发生时间
+	PigBatchID    uint      `json:"pig_batch_id"`   // 关联的猪群ID
+	PenID         uint      `json:"pen_id"`         // 发生变动的猪栏ID
+	Quantity      int       `json:"quantity"`       // 变动数量（正数表示增加，负数表示减少）
+	Type          string    `json:"type"`           // 变动类型 (e.g., "群内调栏", "跨群调栏", "销售", "死亡", "新购入")
+	CorrelationID string    `json:"correlation_id"` // 用于关联一次完整操作（如一次调栏会产生两条日志）
+}

internal/infra/repository/pen_repository.go

@@ -0,0 +1,46 @@
+package repository
+
+import (
+	"git.huangwc.com/pig/pig-farm-controller/internal/infra/models"
+	"gorm.io/gorm"
+)
+
+// PenRepository 定义了与猪栏模型相关的数据库操作接口。
+type PenRepository interface {
+	GetPenByIDTx(tx *gorm.DB, penID uint) (*models.Pen, error)
+	GetPensByBatchIDTx(tx *gorm.DB, batchID uint) ([]*models.Pen, error)
+	UpdatePenFieldsTx(tx *gorm.DB, penID uint, updates map[string]interface{}) error
+}
+
+// penRepository 是 PenRepository 接口的 gorm 实现。
+type penRepository struct {
+	db *gorm.DB
+}
+
+// NewPenRepository 创建一个新的 PenRepository 实例。
+func NewPenRepository(db *gorm.DB) PenRepository {
+	return &penRepository{db: db}
+}
+
+// GetPenByIDTx 在指定的事务中，通过ID获取单个猪栏信息。
+func (r *penRepository) GetPenByIDTx(tx *gorm.DB, penID uint) (*models.Pen, error) {
+	var pen models.Pen
+	if err := tx.First(&pen, penID).Error; err != nil {
+		return nil, err
+	}
+	return &pen, nil
+}
+
+// GetPensByBatchIDTx 在指定的事务中，获取一个猪群当前关联的所有猪栏。
+func (r *penRepository) GetPensByBatchIDTx(tx *gorm.DB, batchID uint) ([]*models.Pen, error) {
+	var pens []*models.Pen
+	if err := tx.Where("pig_batch_id = ?", batchID).Find(&pens).Error; err != nil {
+		return nil, err
+	}
+	return pens, nil
+}
+
+// UpdatePenFieldsTx 在指定的事务中，更新一个猪栏的指定字段。
+func (r *penRepository) UpdatePenFieldsTx(tx *gorm.DB, penID uint, updates map[string]interface{}) error {
+	return tx.Model(&models.Pen{}).Where("id = ?", penID).Updates(updates).Error
+}

internal/infra/repository/pig_batch_repository.go

@@ -9,6 +9,7 @@ import (
 type PigBatchRepository interface {
 	CreatePigBatch(batch *models.PigBatch) (*models.PigBatch, error)
 	GetPigBatchByID(id uint) (*models.PigBatch, error)
+	GetPigBatchByIDTx(tx *gorm.DB, id uint) (*models.PigBatch, error)
 	// UpdatePigBatch 更新一个猪批次，返回更新后的批次、受影响的行数和错误
 	UpdatePigBatch(batch *models.PigBatch) (*models.PigBatch, int64, error)
 	// DeletePigBatch 根据ID删除一个猪批次，返回受影响的行数和错误
@@ -83,3 +84,12 @@ func (r *gormPigBatchRepository) ListPigBatches(isActive *bool) ([]*models.PigBa
 	}
 	return batches, nil
 }
+
+// GetPigBatchByIDTx 在指定的事务中，通过ID获取单个猪批次
+func (r *gormPigBatchRepository) GetPigBatchByIDTx(tx *gorm.DB, id uint) (*models.PigBatch, error) {
+	var batch models.PigBatch
+	if err := tx.First(&batch, id).Error; err != nil {
+		return nil, err
+	}
+	return &batch, nil
+}