jacob-mathison/query-orchestration
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Activity

batch upload impl

Browse Source
This commit is contained in:
jay brown
2025-08-06 15:06:18 -07:00
parent fac5615c15
commit 7014fa790b
17 changed files with 1272 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/database/README.md
+63 -10
View File
@@ -6,6 +6,46 @@ Package database provides the data access layer for the query orchestration plat
## Architecture
### Service Layer Architecture
The database access follows a clean, layered architecture pattern:
1. **Repository Layer** (`repository/`)
- SQLC-generated code providing type-safe database queries
- `Queries` struct contains all database operations
- Generated from SQL files in `queries/` directory
- Accessed throughout codebase via `ConfigProvider.GetDBQueries()`
2. **Service Layer** (`internal/*/service.go`)
- Each domain has its own service containing business logic
- Services use the repository layer for all database access
- Examples: `client/service.go`, `document/service.go`, `query/service.go`
3. **Configuration Layer** (`internal/serviceconfig/database/`)
- Manages database connections and connection pooling
- Provides transaction support via `ExecuteDBTransaction()`
- Handles migration execution on startup
4. **Database Access Pattern**
```
API Controllers → Service Layer (business logic)
↓
ConfigProvider.GetDBQueries()
↓
Repository Layer (SQLC-generated)
↓
PostgreSQL Database
```
### Architecture Principles
- **No Direct SQL in Business Logic**: All SQL queries are defined in `.sql` files
- **Type Safety**: SQLC generates Go types from SQL schemas
- **Clean Separation**: HTTP handlers → Services → Repository → Database
- **Transaction Support**: Coordinated through ConfigProvider interface
- **Code Generation**: Repository code is generated, not hand-written
- **Single Source of Truth**: Database schema defined in migrations
### Core Components
#### Database Schema
@@ -16,6 +56,7 @@ The database schema supports document processing and query orchestration with th
- **Collectors**: Query collection specifications per client
- **Results**: Processed query outputs linked to documents
- **Text Extractions**: OCR results from document processing
- **Batch Uploads**: Batch document upload tracking with progress monitoring
#### Migration System
- Uses `golang-migrate` for schema version management
@@ -53,6 +94,7 @@ The database schema supports document processing and query orchestration with th
- **results**: Query execution results per document
- **collectors**: Client-specific query collections
- **collectorVersions**: Collector version history
- **batch_uploads**: Batch document upload tracking with progress and failure metrics
### Views
@@ -83,21 +125,20 @@ The database schema supports document processing and query orchestration with th
// Create database configuration
dbConfig := &serviceconfig.DBConfig{
DBHost: "localhost",
DBPort: "5432",
DBPort: 5432,
DBName: "querydb",
DBUser: "dbuser",
DBPassword: "dbpass",
}
// Get connection pool
pool, err := dbConfig.GetDBPool(ctx)
// Initialize connection pool
err := dbConfig.SetDBPoolConfig()
if err != nil {
return err
}
defer pool.Close()
// Create repository queries instance
queries := repository.New(pool)
// Get repository queries instance
queries := dbConfig.GetDBQueries()
```
### Transaction Management
@@ -168,6 +209,15 @@ err := migrator.Up()
## SQLC Integration
### Configuration
SQLC is configured via `sqlc.yml` in the project root with:
- PostgreSQL engine with pgx/v5 driver
- Queries sourced from `internal/database/queries/`
- Schema from `internal/database/migrations/`
- Generated code output to `internal/database/repository/`
- Custom rules enforcing query performance and safety
- UUID type mapping to `github.com/google/uuid`
### Query Definition Format
```sql
@@ -187,12 +237,15 @@ SET status = $2, updated = CURRENT_TIMESTAMP
WHERE id = $1;
```
### Generated Code
### Generated Code Features
SQLC generates:
- Type-safe query functions
- Type-safe query functions with parameter structs
- Struct definitions matching table schemas
- Null handling with sql.NullString, sql.NullTime, etc.
- Parameter validation
- Null handling with pointers for nullable fields
- UUID type support via google/uuid package
- SQL comments included in generated code
- Enum validation methods
- Empty slice initialization for array returns
## Testing
internal/database/migrations/00000000000104_add_batch_storage_columns.down.sql
+5
View File
@@ -0,0 +1,5 @@
-- Remove storage columns and constraint
ALTER TABLE batch_uploads DROP CONSTRAINT IF EXISTS batch_storage_required;
ALTER TABLE batch_uploads DROP COLUMN IF EXISTS file_size_bytes;
ALTER TABLE batch_uploads DROP COLUMN IF EXISTS archive_key;
ALTER TABLE batch_uploads DROP COLUMN IF EXISTS archive_bucket;
internal/database/migrations/00000000000104_add_batch_storage_columns.up.sql
+11
View File
@@ -0,0 +1,11 @@
-- Add storage metadata columns to batch_uploads table
ALTER TABLE batch_uploads ADD COLUMN archive_bucket text;
ALTER TABLE batch_uploads ADD COLUMN archive_key text;
ALTER TABLE batch_uploads ADD COLUMN file_size_bytes bigint;
-- Add constraint to ensure storage metadata is consistent (if any storage field is set, bucket and key must both be set)
ALTER TABLE batch_uploads ADD CONSTRAINT batch_storage_consistent
CHECK (
(archive_bucket IS NULL AND archive_key IS NULL AND file_size_bytes IS NULL) OR
(archive_bucket IS NOT NULL AND archive_key IS NOT NULL)
);
internal/database/queries/batch.sql
+20
View File
@@ -2,6 +2,18 @@
INSERT INTO batch_uploads (client_id, original_filename, total_documents)
VALUES ($1, $2, $3) RETURNING id;
-- name: CreateBatchUploadWithStorage :one
INSERT INTO batch_uploads (
client_id,
original_filename,
total_documents,
status,
archive_bucket,
archive_key,
file_size_bytes
) VALUES ($1, $2, $3, $4::batch_status, $5, $6, $7)
RETURNING id, created_at;
-- name: GetBatchUpload :one
SELECT id, client_id, original_filename, total_documents, processed_documents,
failed_documents, invalid_type_documents, status, progress_percent,
@@ -9,6 +21,14 @@ SELECT id, client_id, original_filename, total_documents, processed_documents,
FROM batch_uploads
WHERE id = $1 AND client_id = $2;
-- name: GetBatchUploadWithStorage :one
SELECT id, client_id, original_filename, total_documents,
processed_documents, failed_documents, invalid_type_documents,
status, archive_bucket, archive_key, file_size_bytes,
failed_filenames, created_at, completed_at
FROM batch_uploads
WHERE id = $1 AND client_id = $2;
-- name: ListBatchUploads :many
SELECT id, client_id, original_filename, total_documents, processed_documents,
failed_documents, invalid_type_documents, status, progress_percent,
internal/database/repository/batch.sql.go
+133 -2
View File
@@ -96,6 +96,61 @@ func (q *Queries) CreateBatchUpload(ctx context.Context, arg *CreateBatchUploadP
return id, err
}
const createBatchUploadWithStorage = `-- name: CreateBatchUploadWithStorage :one
INSERT INTO batch_uploads (
client_id,
original_filename,
total_documents,
status,
archive_bucket,
archive_key,
file_size_bytes
) VALUES ($1, $2, $3, $4::batch_status, $5, $6, $7)
RETURNING id, created_at
`
type CreateBatchUploadWithStorageParams struct {
ClientID string `db:"client_id"`
OriginalFilename string `db:"original_filename"`
TotalDocuments int32 `db:"total_documents"`
Column4 BatchStatus `db:"column_4"`
ArchiveBucket *string `db:"archive_bucket"`
ArchiveKey *string `db:"archive_key"`
FileSizeBytes *int64 `db:"file_size_bytes"`
}
type CreateBatchUploadWithStorageRow struct {
ID uuid.UUID `db:"id"`
CreatedAt pgtype.Timestamp `db:"created_at"`
}
// CreateBatchUploadWithStorage
//
// INSERT INTO batch_uploads (
// client_id,
// original_filename,
// total_documents,
// status,
// archive_bucket,
// archive_key,
// file_size_bytes
// ) VALUES ($1, $2, $3, $4::batch_status, $5, $6, $7)
// RETURNING id, created_at
func (q *Queries) CreateBatchUploadWithStorage(ctx context.Context, arg *CreateBatchUploadWithStorageParams) (*CreateBatchUploadWithStorageRow, error) {
row := q.db.QueryRow(ctx, createBatchUploadWithStorage,
arg.ClientID,
arg.OriginalFilename,
arg.TotalDocuments,
arg.Column4,
arg.ArchiveBucket,
arg.ArchiveKey,
arg.FileSizeBytes,
)
var i CreateBatchUploadWithStorageRow
err := row.Scan(&i.ID, &i.CreatedAt)
return &i, err
}
const getBatchUpload = `-- name: GetBatchUpload :one
SELECT id, client_id, original_filename, total_documents, processed_documents,
failed_documents, invalid_type_documents, status, progress_percent,
@@ -109,6 +164,21 @@ type GetBatchUploadParams struct {
ClientID string `db:"client_id"`
}
type GetBatchUploadRow struct {
ID uuid.UUID `db:"id"`
ClientID string `db:"client_id"`
OriginalFilename string `db:"original_filename"`
TotalDocuments int32 `db:"total_documents"`
ProcessedDocuments int32 `db:"processed_documents"`
FailedDocuments int32 `db:"failed_documents"`
InvalidTypeDocuments int32 `db:"invalid_type_documents"`
Status BatchStatus `db:"status"`
ProgressPercent int32 `db:"progress_percent"`
FailedFilenames []byte `db:"failed_filenames"`
CreatedAt pgtype.Timestamp `db:"created_at"`
CompletedAt pgtype.Timestamp `db:"completed_at"`
}
// GetBatchUpload
//
// SELECT id, client_id, original_filename, total_documents, processed_documents,
@@ -116,9 +186,9 @@ type GetBatchUploadParams struct {
// failed_filenames, created_at, completed_at
// FROM batch_uploads
// WHERE id = $1 AND client_id = $2
func (q *Queries) GetBatchUpload(ctx context.Context, arg *GetBatchUploadParams) (*BatchUpload, error) {
func (q *Queries) GetBatchUpload(ctx context.Context, arg *GetBatchUploadParams) (*GetBatchUploadRow, error) {
row := q.db.QueryRow(ctx, getBatchUpload, arg.ID, arg.ClientID)
var i BatchUpload
var i GetBatchUploadRow
err := row.Scan(
&i.ID,
&i.ClientID,
@@ -136,6 +206,67 @@ func (q *Queries) GetBatchUpload(ctx context.Context, arg *GetBatchUploadParams)
return &i, err
}
const getBatchUploadWithStorage = `-- name: GetBatchUploadWithStorage :one
SELECT id, client_id, original_filename, total_documents,
processed_documents, failed_documents, invalid_type_documents,
status, archive_bucket, archive_key, file_size_bytes,
failed_filenames, created_at, completed_at
FROM batch_uploads
WHERE id = $1 AND client_id = $2
`
type GetBatchUploadWithStorageParams struct {
ID uuid.UUID `db:"id"`
ClientID string `db:"client_id"`
}
type GetBatchUploadWithStorageRow struct {
ID uuid.UUID `db:"id"`
ClientID string `db:"client_id"`
OriginalFilename string `db:"original_filename"`
TotalDocuments int32 `db:"total_documents"`
ProcessedDocuments int32 `db:"processed_documents"`
FailedDocuments int32 `db:"failed_documents"`
InvalidTypeDocuments int32 `db:"invalid_type_documents"`
Status BatchStatus `db:"status"`
ArchiveBucket *string `db:"archive_bucket"`
ArchiveKey *string `db:"archive_key"`
FileSizeBytes *int64 `db:"file_size_bytes"`
FailedFilenames []byte `db:"failed_filenames"`
CreatedAt pgtype.Timestamp `db:"created_at"`
CompletedAt pgtype.Timestamp `db:"completed_at"`
}
// GetBatchUploadWithStorage
//
// SELECT id, client_id, original_filename, total_documents,
// processed_documents, failed_documents, invalid_type_documents,
// status, archive_bucket, archive_key, file_size_bytes,
// failed_filenames, created_at, completed_at
// FROM batch_uploads
// WHERE id = $1 AND client_id = $2
func (q *Queries) GetBatchUploadWithStorage(ctx context.Context, arg *GetBatchUploadWithStorageParams) (*GetBatchUploadWithStorageRow, error) {
row := q.db.QueryRow(ctx, getBatchUploadWithStorage, arg.ID, arg.ClientID)
var i GetBatchUploadWithStorageRow
err := row.Scan(
&i.ID,
&i.ClientID,
&i.OriginalFilename,
&i.TotalDocuments,
&i.ProcessedDocuments,
&i.FailedDocuments,
&i.InvalidTypeDocuments,
&i.Status,
&i.ArchiveBucket,
&i.ArchiveKey,
&i.FileSizeBytes,
&i.FailedFilenames,
&i.CreatedAt,
&i.CompletedAt,
)
return &i, err
}
const getDocumentsByBatchId = `-- name: GetDocumentsByBatchId :many
SELECT id, clientId, hash
FROM documents
internal/database/repository/models.go
+3
View File
@@ -245,6 +245,9 @@ type BatchUpload struct {
FailedFilenames []byte `db:"failed_filenames"`
CreatedAt pgtype.Timestamp `db:"created_at"`
CompletedAt pgtype.Timestamp `db:"completed_at"`
ArchiveBucket *string `db:"archive_bucket"`
ArchiveKey *string `db:"archive_key"`
FileSizeBytes *int64 `db:"file_size_bytes"`
}
type Client struct {