swagger part 1
This commit is contained in:
jay brown
@@ -0,0 +1,28 @@
|
||||
# Document Batch Processing
|
||||
|
||||
This directory contains the service layer for managing document batch uploads.
|
||||
|
||||
## Purpose
|
||||
|
||||
The batch service provides a business logic layer between the HTTP API handlers and the database repository for batch upload operations. It handles:
|
||||
|
||||
- **Model Conversion**: Translates between OpenAPI-generated types and internal database models
|
||||
- **Business Logic**: Encapsulates batch-specific operations (CRUD, progress tracking)
|
||||
- **Data Validation**: Manages UUID conversions, JSONB parsing, and nullable type handling
|
||||
|
||||
## Architecture
|
||||
|
||||
```
|
||||
HTTP Request → API Handler → Batch Service → Database Repository → PostgreSQL
|
||||
```
|
||||
|
||||
The service is **stateless** and **synchronous** - it does not run background processes. Actual document processing happens through the existing Runner pipeline (`storeEventRunner` → `docInitRunner` → etc.).
|
||||
|
||||
## Files
|
||||
|
||||
- `service.go` - Main batch service implementation with CRUD operations
|
||||
- `service_test.go` - Unit tests using testcontainers for database integration
|
||||
|
||||
## Usage
|
||||
|
||||
The service is instantiated by API handlers in `/api/queryAPI/documents.go` to manage batch upload metadata, progress tracking, and status updates.
|
||||
@@ -0,0 +1,218 @@
|
||||
package batch
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"queryorchestration/internal/database/repository"
|
||||
"queryorchestration/internal/serviceconfig"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/jackc/pgx/v5/pgtype"
|
||||
)
|
||||
|
||||
// BatchUploadSummary represents a summary of a batch upload
|
||||
type BatchUploadSummary struct {
|
||||
ID uuid.UUID `json:"batch_id"`
|
||||
ClientID string `json:"client_id"`
|
||||
OriginalFilename string `json:"original_filename"`
|
||||
Status string `json:"status"`
|
||||
TotalDocuments int32 `json:"total_documents"`
|
||||
ProcessedDocuments int32 `json:"processed_documents"`
|
||||
FailedDocuments int32 `json:"failed_documents"`
|
||||
InvalidTypeDocuments int32 `json:"invalid_type_documents"`
|
||||
ProgressPercent int32 `json:"progress_percent"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
CompletedAt *time.Time `json:"completed_at,omitempty"`
|
||||
}
|
||||
|
||||
// BatchUploadDetails includes full details with failed filenames
|
||||
type BatchUploadDetails struct {
|
||||
BatchUploadSummary
|
||||
FailedFilenames []string `json:"failed_filenames"`
|
||||
}
|
||||
|
||||
// Service handles batch upload operations
|
||||
type Service struct {
|
||||
cfg serviceconfig.ConfigProvider
|
||||
}
|
||||
|
||||
// New creates a new batch service
|
||||
func New(cfg serviceconfig.ConfigProvider) *Service {
|
||||
return &Service{
|
||||
cfg: cfg,
|
||||
}
|
||||
}
|
||||
|
||||
// Create creates a new batch upload record
|
||||
func (s *Service) Create(ctx context.Context, clientID string, filename string, totalDocs int32) (uuid.UUID, error) {
|
||||
return s.cfg.GetDBQueries().CreateBatchUpload(ctx, &repository.CreateBatchUploadParams{
|
||||
ClientID: clientID,
|
||||
OriginalFilename: filename,
|
||||
TotalDocuments: totalDocs,
|
||||
})
|
||||
}
|
||||
|
||||
// Get retrieves batch upload details
|
||||
func (s *Service) Get(ctx context.Context, clientID string, batchID uuid.UUID) (*BatchUploadDetails, error) {
|
||||
batch, err := s.cfg.GetDBQueries().GetBatchUpload(ctx, &repository.GetBatchUploadParams{
|
||||
ID: batchID,
|
||||
ClientID: clientID,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return s.convertToDetails(batch), nil
|
||||
}
|
||||
|
||||
// List retrieves all batch uploads for a client with pagination
|
||||
func (s *Service) List(ctx context.Context, clientID string, limit int32, offset int32) ([]*BatchUploadSummary, error) {
|
||||
if limit <= 0 {
|
||||
limit = 20 // default limit
|
||||
}
|
||||
if limit > 100 {
|
||||
limit = 100 // max limit
|
||||
}
|
||||
|
||||
batches, err := s.cfg.GetDBQueries().ListBatchUploads(ctx, &repository.ListBatchUploadsParams{
|
||||
ClientID: clientID,
|
||||
Limit: int64(limit),
|
||||
Offset: int64(offset),
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
summaries := make([]*BatchUploadSummary, len(batches))
|
||||
for i, batch := range batches {
|
||||
summaries[i] = s.convertToSummary(batch)
|
||||
}
|
||||
|
||||
return summaries, nil
|
||||
}
|
||||
|
||||
// Cancel cancels a batch upload in progress
|
||||
func (s *Service) Cancel(ctx context.Context, clientID string, batchID uuid.UUID) error {
|
||||
return s.cfg.GetDBQueries().CancelBatchUpload(ctx, &repository.CancelBatchUploadParams{
|
||||
ID: batchID,
|
||||
ClientID: clientID,
|
||||
})
|
||||
}
|
||||
|
||||
// UpdateProgress updates batch processing progress
|
||||
func (s *Service) UpdateProgress(ctx context.Context, clientID string, batchID uuid.UUID, processed, failed, invalidType int32) error {
|
||||
// Calculate progress percentage
|
||||
batch, err := s.cfg.GetDBQueries().GetBatchUpload(ctx, &repository.GetBatchUploadParams{
|
||||
ID: batchID,
|
||||
ClientID: clientID,
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to get batch for progress update: %w", err)
|
||||
}
|
||||
|
||||
total := batch.TotalDocuments
|
||||
if total == 0 {
|
||||
return fmt.Errorf("batch has zero total documents")
|
||||
}
|
||||
|
||||
progressPercent := int32(float64(processed+failed+invalidType) / float64(total) * 100)
|
||||
|
||||
return s.cfg.GetDBQueries().UpdateBatchProgress(ctx, &repository.UpdateBatchProgressParams{
|
||||
ID: batchID,
|
||||
ProcessedDocuments: processed,
|
||||
FailedDocuments: failed,
|
||||
InvalidTypeDocuments: invalidType,
|
||||
ProgressPercent: progressPercent,
|
||||
})
|
||||
}
|
||||
|
||||
// MarkCompleted marks a batch as completed
|
||||
func (s *Service) MarkCompleted(ctx context.Context, batchID uuid.UUID) error {
|
||||
return s.cfg.GetDBQueries().UpdateBatchStatus(ctx, &repository.UpdateBatchStatusParams{
|
||||
ID: batchID,
|
||||
Column2: repository.BatchStatusCompleted,
|
||||
})
|
||||
}
|
||||
|
||||
// MarkFailed marks a batch as failed
|
||||
func (s *Service) MarkFailed(ctx context.Context, batchID uuid.UUID) error {
|
||||
return s.cfg.GetDBQueries().UpdateBatchStatus(ctx, &repository.UpdateBatchStatusParams{
|
||||
ID: batchID,
|
||||
Column2: repository.BatchStatusFailed,
|
||||
})
|
||||
}
|
||||
|
||||
// AddFailedFilename adds a failed filename to the batch
|
||||
func (s *Service) AddFailedFilename(ctx context.Context, batchID uuid.UUID, filename string) error {
|
||||
return s.cfg.GetDBQueries().AddFailedFilename(ctx, &repository.AddFailedFilenameParams{
|
||||
ID: batchID,
|
||||
Column2: filename,
|
||||
})
|
||||
}
|
||||
|
||||
// convertToDetails converts repository model to service details model
|
||||
func (s *Service) convertToDetails(batch *repository.BatchUpload) *BatchUploadDetails {
|
||||
details := &BatchUploadDetails{
|
||||
BatchUploadSummary: BatchUploadSummary{
|
||||
ID: batch.ID,
|
||||
ClientID: batch.ClientID,
|
||||
OriginalFilename: batch.OriginalFilename,
|
||||
Status: string(batch.Status),
|
||||
TotalDocuments: batch.TotalDocuments,
|
||||
ProcessedDocuments: batch.ProcessedDocuments,
|
||||
FailedDocuments: batch.FailedDocuments,
|
||||
InvalidTypeDocuments: batch.InvalidTypeDocuments,
|
||||
ProgressPercent: batch.ProgressPercent,
|
||||
CreatedAt: batch.CreatedAt.Time,
|
||||
},
|
||||
FailedFilenames: make([]string, 0),
|
||||
}
|
||||
|
||||
if batch.CompletedAt.Valid {
|
||||
details.CompletedAt = &batch.CompletedAt.Time
|
||||
}
|
||||
|
||||
// Parse failed filenames from JSONB
|
||||
if len(batch.FailedFilenames) > 0 {
|
||||
var filenames []string
|
||||
if err := json.Unmarshal(batch.FailedFilenames, &filenames); err == nil {
|
||||
details.FailedFilenames = filenames
|
||||
}
|
||||
}
|
||||
|
||||
return details
|
||||
}
|
||||
|
||||
// convertToSummary converts repository list row to service summary model
|
||||
func (s *Service) convertToSummary(batch *repository.ListBatchUploadsRow) *BatchUploadSummary {
|
||||
summary := &BatchUploadSummary{
|
||||
ID: batch.ID,
|
||||
ClientID: batch.ClientID,
|
||||
OriginalFilename: batch.OriginalFilename,
|
||||
Status: string(batch.Status),
|
||||
TotalDocuments: batch.TotalDocuments,
|
||||
ProcessedDocuments: batch.ProcessedDocuments,
|
||||
FailedDocuments: batch.FailedDocuments,
|
||||
InvalidTypeDocuments: batch.InvalidTypeDocuments,
|
||||
ProgressPercent: batch.ProgressPercent,
|
||||
CreatedAt: s.pgTimestampToTime(batch.CreatedAt),
|
||||
}
|
||||
|
||||
if batch.CompletedAt.Valid {
|
||||
t := s.pgTimestampToTime(batch.CompletedAt)
|
||||
summary.CompletedAt = &t
|
||||
}
|
||||
|
||||
return summary
|
||||
}
|
||||
|
||||
// pgTimestampToTime converts pgtype.Timestamp to time.Time
|
||||
func (s *Service) pgTimestampToTime(ts pgtype.Timestamp) time.Time {
|
||||
if ts.Valid {
|
||||
return ts.Time
|
||||
}
|
||||
return time.Time{}
|
||||
}
|
||||
@@ -0,0 +1,185 @@
|
||||
package batch_test
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"queryorchestration/internal/database/repository"
|
||||
documentbatch "queryorchestration/internal/document/batch"
|
||||
"queryorchestration/internal/serviceconfig"
|
||||
"queryorchestration/internal/test"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
type TestConfig struct {
|
||||
serviceconfig.BaseConfig
|
||||
}
|
||||
|
||||
func TestUpdateProgress(t *testing.T) {
|
||||
cfg := &TestConfig{}
|
||||
test.CreateDB(t, cfg)
|
||||
|
||||
service := documentbatch.New(cfg)
|
||||
|
||||
// Create test client
|
||||
clientID := "test_client_progress"
|
||||
err := cfg.GetDBQueries().CreateClient(t.Context(), &repository.CreateClientParams{
|
||||
Clientid: clientID,
|
||||
Name: "Test Client Progress",
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
// Create test batch with total documents
|
||||
batchID, err := cfg.GetDBQueries().CreateBatchUpload(t.Context(), &repository.CreateBatchUploadParams{
|
||||
ClientID: clientID,
|
||||
OriginalFilename: "test.zip",
|
||||
TotalDocuments: 10,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
// Test updating progress
|
||||
err = service.UpdateProgress(t.Context(), clientID, batchID, 5, 2, 1)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Verify progress was updated
|
||||
batch, err := cfg.GetDBQueries().GetBatchUpload(t.Context(), &repository.GetBatchUploadParams{
|
||||
ID: batchID,
|
||||
ClientID: clientID,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, int32(5), batch.ProcessedDocuments)
|
||||
assert.Equal(t, int32(2), batch.FailedDocuments)
|
||||
assert.Equal(t, int32(1), batch.InvalidTypeDocuments)
|
||||
assert.Equal(t, int32(80), batch.ProgressPercent) // (5+2+1)/10 * 100 = 80%
|
||||
}
|
||||
|
||||
func TestUpdateProgressZeroTotalDocuments(t *testing.T) {
|
||||
cfg := &TestConfig{}
|
||||
test.CreateDB(t, cfg)
|
||||
|
||||
service := documentbatch.New(cfg)
|
||||
|
||||
// Create test client
|
||||
clientID := "test_client_zero"
|
||||
err := cfg.GetDBQueries().CreateClient(t.Context(), &repository.CreateClientParams{
|
||||
Clientid: clientID,
|
||||
Name: "Test Client Zero",
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
// Create test batch with zero total documents
|
||||
batchID, err := cfg.GetDBQueries().CreateBatchUpload(t.Context(), &repository.CreateBatchUploadParams{
|
||||
ClientID: clientID,
|
||||
OriginalFilename: "test.zip",
|
||||
TotalDocuments: 0,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
// Test updating progress should fail with zero total documents
|
||||
err = service.UpdateProgress(t.Context(), clientID, batchID, 5, 2, 1)
|
||||
require.Error(t, err)
|
||||
assert.Contains(t, err.Error(), "batch has zero total documents")
|
||||
}
|
||||
|
||||
func TestMarkCompleted(t *testing.T) {
|
||||
cfg := &TestConfig{}
|
||||
test.CreateDB(t, cfg)
|
||||
|
||||
service := documentbatch.New(cfg)
|
||||
|
||||
// Create test client
|
||||
clientID := "test_client_complete"
|
||||
err := cfg.GetDBQueries().CreateClient(t.Context(), &repository.CreateClientParams{
|
||||
Clientid: clientID,
|
||||
Name: "Test Client Complete",
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
// Create test batch
|
||||
batchID, err := cfg.GetDBQueries().CreateBatchUpload(t.Context(), &repository.CreateBatchUploadParams{
|
||||
ClientID: clientID,
|
||||
OriginalFilename: "test.zip",
|
||||
TotalDocuments: 10,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
// Test marking as completed
|
||||
err = service.MarkCompleted(t.Context(), batchID)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Verify status was updated
|
||||
batch, err := cfg.GetDBQueries().GetBatchUpload(t.Context(), &repository.GetBatchUploadParams{
|
||||
ID: batchID,
|
||||
ClientID: clientID,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, repository.BatchStatusCompleted, batch.Status)
|
||||
}
|
||||
|
||||
func TestMarkFailed(t *testing.T) {
|
||||
cfg := &TestConfig{}
|
||||
test.CreateDB(t, cfg)
|
||||
|
||||
service := documentbatch.New(cfg)
|
||||
|
||||
// Create test client
|
||||
clientID := "test_client_failed"
|
||||
err := cfg.GetDBQueries().CreateClient(t.Context(), &repository.CreateClientParams{
|
||||
Clientid: clientID,
|
||||
Name: "Test Client Failed",
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
// Create test batch
|
||||
batchID, err := cfg.GetDBQueries().CreateBatchUpload(t.Context(), &repository.CreateBatchUploadParams{
|
||||
ClientID: clientID,
|
||||
OriginalFilename: "test.zip",
|
||||
TotalDocuments: 10,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
// Test marking as failed
|
||||
err = service.MarkFailed(t.Context(), batchID)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Verify status was updated
|
||||
batch, err := cfg.GetDBQueries().GetBatchUpload(t.Context(), &repository.GetBatchUploadParams{
|
||||
ID: batchID,
|
||||
ClientID: clientID,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, repository.BatchStatusFailed, batch.Status)
|
||||
}
|
||||
|
||||
func TestAddFailedFilename(t *testing.T) {
|
||||
cfg := &TestConfig{}
|
||||
test.CreateDB(t, cfg)
|
||||
|
||||
service := documentbatch.New(cfg)
|
||||
|
||||
// Create test client
|
||||
clientID := "test_client_filename"
|
||||
err := cfg.GetDBQueries().CreateClient(t.Context(), &repository.CreateClientParams{
|
||||
Clientid: clientID,
|
||||
Name: "Test Client Filename",
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
// Create test batch
|
||||
batchID, err := cfg.GetDBQueries().CreateBatchUpload(t.Context(), &repository.CreateBatchUploadParams{
|
||||
ClientID: clientID,
|
||||
OriginalFilename: "test.zip",
|
||||
TotalDocuments: 10,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
// Test adding failed filename
|
||||
err = service.AddFailedFilename(t.Context(), batchID, "failed_file.pdf")
|
||||
require.NoError(t, err)
|
||||
|
||||
// Get the full details to check failed filenames
|
||||
details, err := service.Get(t.Context(), clientID, batchID)
|
||||
require.NoError(t, err)
|
||||
assert.Contains(t, details.FailedFilenames, "failed_file.pdf")
|
||||
}
|
||||
Reference in New Issue
Block a user