Merged in feature/permit-policy-cleanup (pull request #210)

cleanup permit policies and correct documentation

* docs

* policy cleanup

* refactor

* Merge remote-tracking branch 'origin/feature/permit-policy-cleanup' into feature/permit-policy-cleanup

* docs and edits
This commit is contained in:
Jay Brown
2026-02-19 20:22:59 +00:00
parent 58912a66d4
commit 6dccf494f8
70 changed files with 785 additions and 2221 deletions
+4
View File
@@ -80,3 +80,7 @@ Thumbs.db
**/.claude/settings.local.json
deployments/permit.env
# local agent journals
journals/
requirements/
+73
View File
@@ -0,0 +1,73 @@
#!/usr/bin/env bash
set -euo pipefail
# Build Linux binaries for the cognito auth harness main.go
# Usage examples:
# ./build_linux.sh # builds linux/amd64 -> cognito.auth.harness_linux_amd64
# ./build_linux.sh --arch arm64 # builds linux/arm64 -> cognito.auth.harness_linux_arm64
# ./build_linux.sh --all # builds both amd64 and arm64
# ./build_linux.sh ./my_out # custom output for single-arch build
#
# Notes:
# - For AWS Graviton instances use --arch arm64
# - For AWS CloudShell or x86 EC2 use default (amd64)
#
# Env overrides:
# GOARCH: target architecture (amd64|arm64) when not using --all
# CGO_ENABLED: 0 by default for cross-compilation
SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" &>/dev/null && pwd)"
cd "$SCRIPT_DIR"
GOOS=linux
DEFAULT_ARCH="${GOARCH:-amd64}"
CGO_ENABLED="${CGO_ENABLED:-0}"
OUT_OVERRIDE=""
ARCH_ARG=""
BUILD_BOTH=false
print_help() {
sed -n '1,35p' "$0" | sed 's/^# \{0,1\}//'
}
while [[ $# -gt 0 ]]; do
case "$1" in
--arch)
ARCH_ARG="$2"; shift 2 ;;
--all)
BUILD_BOTH=true; shift ;;
-h|--help)
print_help; exit 0 ;;
--)
shift; break ;;
-* )
echo "Unknown option: $1" >&2; exit 1 ;;
* )
# positional: output override for single-arch build
OUT_OVERRIDE="$1"; shift ;;
esac
done
build_one() {
local arch="$1"
local out_path="$2"
echo "Building Linux binary (GOARCH=${arch})"
echo " GOOS=${GOOS} GOARCH=${arch} CGO_ENABLED=${CGO_ENABLED}"
echo " Output: ${out_path}"
GOOS="$GOOS" GOARCH="$arch" CGO_ENABLED="$CGO_ENABLED" \
go build -o "$out_path" main.go
echo "✅ Built ${out_path}"
}
if $BUILD_BOTH; then
build_one amd64 "cognito.auth.harness_linux_amd64"
build_one arm64 "cognito.auth.harness_linux_arm64"
exit 0
fi
ARCH="${ARCH_ARG:-$DEFAULT_ARCH}"
DEFAULT_OUT="cognito.auth.harness_linux_${ARCH}"
OUT_PATH="${OUT_OVERRIDE:-$DEFAULT_OUT}"
build_one "$ARCH" "$OUT_PATH"
+19
View File
@@ -0,0 +1,19 @@
#!/usr/bin/env bash
set -euo pipefail
# Build Linux AMD64 binary for AWS CloudShell
# Usage: ./build.linux.sh
#
# Output: dist/setup_permit
SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" &>/dev/null && pwd)"
cd "$SCRIPT_DIR"
mkdir -p dist
echo "Building Linux AMD64 binary for AWS CloudShell..."
echo " GOOS=linux GOARCH=amd64 CGO_ENABLED=0"
GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build -o dist/setup_permit .
echo "Built: dist/setup_permit"
+35
View File
@@ -0,0 +1,35 @@
#!/bin/bash
# Delete the default Permit.io roles that conflict with our resource names.
# Only run this on a brand new environment since it creates conflicting roles to start that need
# to be deleted.
export PERMIT_API_KEY="permit_key_omitted"
PROJECT_ID="default"
ENV_ID="production"
echo "Deleting conflicting default roles from Production environment..."
echo ""
# Delete the 'admin' role
echo "Deleting 'admin' role..."
curl -X DELETE "https://api.permit.io/v2/schema/${PROJECT_ID}/${ENV_ID}/roles/admin" \
-H "Authorization: Bearer ${PERMIT_API_KEY}" \
-H "Content-Type: application/json"
echo ""
# Delete the 'editor' role (doesn't conflict but keeping consistency)
echo "Deleting 'editor' role..."
curl -X DELETE "https://api.permit.io/v2/schema/${PROJECT_ID}/${ENV_ID}/roles/editor" \
-H "Authorization: Bearer ${PERMIT_API_KEY}" \
-H "Content-Type: application/json"
echo ""
# Delete the 'viewer' role (doesn't conflict but keeping consistency)
echo "Deleting 'viewer' role..."
curl -X DELETE "https://api.permit.io/v2/schema/${PROJECT_ID}/${ENV_ID}/roles/viewer" \
-H "Authorization: Bearer ${PERMIT_API_KEY}" \
-H "Content-Type: application/json"
echo ""
echo "Done! You can now run the setup script."
@@ -30,13 +30,6 @@ resources:
- patch
- delete
- name: query
description: Query definitions and execution
actions:
- get
- post
- patch
- name: document
description: Document upload and management
actions:
@@ -51,7 +44,7 @@ resources:
- post
- name: documents
description: Document search and listing operations
description: Label operations on documents
actions:
- get
- post
@@ -69,8 +62,8 @@ resources:
- post
- patch
- name: home
description: Home page access
- name: labels
description: Label-based document lookup
actions:
- get
@@ -95,10 +88,6 @@ roles:
- post
- patch
- delete
query:
- get
- post
- patch
document:
- get
- post
@@ -116,7 +105,7 @@ roles:
- get
- post
- patch
home:
labels:
- get
- name: user_admin
@@ -137,8 +126,6 @@ roles:
folders:
- get
- post
home:
- get
- name: auditor
description: Full read-only access across the entire system for auditing and compliance
@@ -150,8 +137,6 @@ roles:
- get
clients:
- get
query:
- get
document:
- get
export:
@@ -162,7 +147,7 @@ roles:
- get
folders:
- get
home:
labels:
- get
- name: client_user
@@ -190,7 +175,7 @@ roles:
- get
- post
- patch
home:
labels:
- get
notes: |
This role requires tenant-level filtering by client_id.
@@ -253,19 +238,14 @@ policy_mappings:
- path: /client/{id}/export
methods:
POST: client:post
- path: /client/{id}/folders
methods:
GET: client:get
query_endpoints:
- path: /query
clients_endpoints:
- path: /clients
methods:
GET: query:get
POST: query:post
- path: /query/{id}
methods:
GET: query:get
PATCH: query:patch
- path: /query/{id}/test
methods:
POST: query:post
GET: clients:get
document_endpoints:
- path: /document/{id}
@@ -289,10 +269,64 @@ policy_mappings:
methods:
GET: field-extractions:get
labels_endpoints:
- path: /documents/{documentId}/labels
methods:
POST: documents:post
GET: documents:get
- path: /labels/{labelName}/documents
methods:
GET: labels:get
folders_endpoints:
- path: /folders
methods:
POST: folders:post
- path: /folders/{folderId}
methods:
PATCH: folders:patch
- path: /folders/{folderId}/documents
methods:
GET: folders:get
- path: /folders/{folderId}/metrics
methods:
GET: folders:get
eula_endpoints:
- path: /eula
public: true
- path: /eula/status
authenticated: true
no_permission_check: true
- path: /eula/agree
authenticated: true
no_permission_check: true
- path: /admin/eula
methods:
GET: admin:get
POST: admin:post
- path: /admin/eula/{version_id}
methods:
GET: admin:get
PATCH: admin:patch
- path: /admin/eula/{version_id}/activate
methods:
POST: admin:post
- path: /admin/eula/agreements
methods:
GET: admin:get
- path: /admin/eula/compliance
methods:
GET: admin:get
identity_endpoints:
- path: /identity
authenticated: true
no_permission_check: true
auth_endpoints:
- path: /home
methods:
GET: home:get
public: true
- path: /login
public: true
- path: /login-callback
@@ -300,6 +334,17 @@ policy_mappings:
- path: /logout
authenticated: true
no_permission_check: true
- path: /eula
public: true
- path: /eula/status
authenticated: true
no_permission_check: true
- path: /eula/agree
authenticated: true
no_permission_check: true
- path: /identity
authenticated: true
no_permission_check: true
- path: /swagger/*
public: true
- path: /metrics
@@ -315,14 +360,16 @@ dynamic_mapping:
examples:
- url: /client/AAA
resource: client
- url: /query/019580df-ef65-7676-8de9-94435a93337a
resource: query
- url: /document/019580df-b3f8-7348-9ab1-1e55b3f18ed7
resource: document
- url: /admin/users
resource: admin
- url: /field-extractions
resource: field-extractions
- url: /labels/contract/documents
resource: labels
- url: /folders/019580df-1234-5678-abcd-1e55b3f18ed7/documents
resource: folders
action_extraction:
logic: HTTP method converted to lowercase
@@ -340,7 +387,9 @@ dynamic_mapping:
# Implementation notes
notes:
- Admin endpoints restricted to super_admin and admin roles only
- Admin endpoints restricted to super_admin and user_admin roles only
- Auditor has read-only access across all resources
- Client_user currently has access to all clients (no tenant isolation)
- Tenant isolation will require application changes (not initial requirement)
- /identity, /eula/status, and /eula/agree are auth-only (bypass Permit.io authorization)
- /home and /eula are fully public paths (no auth or permit check)
+11
View File
@@ -0,0 +1,11 @@
# dev environment key
# export PERMIT_API_KEY="permit_key_nwcB5ebXeeZyyt684otN2h70P7EVYWvFjp88QLObZK0uBFLArgPTidmkMxXGGoMh0aSbLTZ8lp9Qg72bYFde1B"
# production key
# export PERMIT_API_KEY="permit_key_UiT5HGXS2kwsWJbZOnzEI86Y2ut4tGqHQcTJoMP6XaqlKmlLFWMVLPwlx0faz35ABigea2htQ6RpGjRDLkLIkT"
go run setup_permit.go -list
# for dev setup (this works)
#go run setup_permit.go -config permit_policies.yaml -project default -env dev
# for production (this does not work)
go run setup_permit.go -config permit_policies.yaml -project default -env production
@@ -2,7 +2,7 @@
All cognito related research code here. Not suitable for production. Only for confirming things
work before placing them in the library.
## cognito_test
## auth_related
This is the harness to demonstrate running all of the cognito auth and RBAC in a simple standalone
main.go. This is standalong documentation for the library contained in `internal/cognitoauth`.
+8 -13
View File
@@ -3,12 +3,11 @@
## DoczyAI Document Processing Platform
### Purpose and Domain
DoczyAI is a cloud-native document processing platform designed to extract structured information from documents (primarily PDFs). The system serves as a multi-tenant platform where organizations can upload documents, perform text extraction, and organize documents with folders and labels for business intelligence and analysis.
DoczyAI is a cloud-native document processing platform for multi-tenant PDF ingestion and processing. Organizations can upload and organize documents with folders and labels, and store structured field extraction results for downstream analysis.
### Core Business Capabilities
- **Multi-tenant Document Processing**: Isolated processing environments per client
- **Batch Document Processing**: ZIP archive uploads for processing multiple documents simultaneously
- **Text Extraction Pipeline**: Automated OCR and text extraction from documents
- **Automated Processing Pipeline**: Seamless flow from document upload to structured output
- **Version Management**: Complete versioning for configurations
- **Export and Integration**: Structured data export capabilities for downstream systems
@@ -20,7 +19,6 @@ DoczyAI is a cloud-native document processing platform designed to extract struc
- **Database**: PostgreSQL 17.2 with SQLC for type-safe queries
- **Messaging**: AWS SQS for event-driven microservices
- **Storage**: AWS S3 for document storage
- **Text Processing**: AWS Textract for OCR and text extraction
- **Authentication**: AWS Cognito with JWT-based authorization
- **Monitoring**: Prometheus metrics with 12 standard metric types and OpenTelemetry tracing
- **Authorization**: Permit.io RBAC integration for fine-grained permissions
@@ -30,7 +28,8 @@ DoczyAI is a cloud-native document processing platform designed to extract struc
### System Architecture Pattern
The system follows an **event-driven microservices architecture** with the following characteristics:
- **8 distinct services**: 1 HTTP API + 6 queue-based runners + 1 utility service
- **Active services**: 1 HTTP API + 5 active queue runners
- **Compatibility services**: Additional deprecated runners may remain in some deployments
- **Asynchronous communication**: SQS queues between services with batch processing capabilities
- **Background Task Framework**: Comprehensive background task execution with periodic scheduling
- **State management**: PostgreSQL as central data store with enhanced schema for batch operations
@@ -55,7 +54,6 @@ graph TB
SE[storeEventRunner<br/>8081] --> DI[docInitRunner<br/>8082]
DI --> DS[docSyncRunner<br/>8083]
DS --> DC[docCleanRunner<br/>8084]
DC --> DT[docTextRunner<br/>8085]
end
subgraph "Synchronization Services"
@@ -66,7 +64,6 @@ graph TB
DB[(PostgreSQL<br/>Database)]
S3[(AWS S3<br/>Storage)]
SQS[AWS SQS<br/>Queues]
TEXTRACT[AWS Textract<br/>Text Extraction]
COGNITO[AWS Cognito<br/>Authentication]
PROM[Prometheus<br/>Monitoring]
end
@@ -83,7 +80,6 @@ graph TB
DI --> DB
DS --> DB
DC --> S3
DT --> TEXTRACT
CS --> SQS
@@ -95,7 +91,7 @@ graph TB
### Infrastructure Requirements
- **Compute**: Container orchestration platform (Docker/Kubernetes)
- **Database**: PostgreSQL 17.2+ with connection pooling
- **AWS Services**: S3, SQS, Textract, Cognito services
- **AWS Services**: S3, SQS, Cognito services
- **Monitoring**: Prometheus-compatible metrics collection
- **Networking**: Load balancer for API endpoints
@@ -117,18 +113,17 @@ graph TB
### External System Dependencies
1. **AWS S3**: Document storage and retrieval
2. **AWS SQS**: Inter-service messaging and event processing
3. **AWS Textract**: OCR and text extraction from documents
4. **AWS Cognito**: User authentication and session management
5. **PostgreSQL**: Central data persistence
3. **AWS Cognito**: User authentication and session management
4. **PostgreSQL**: Central data persistence
### API Integration
- **REST API**: OpenAPI 3.0.3 specification with Swagger UI
- **Authentication**: Bearer token or OAuth2 flows
- **Rate Limiting**: 100 requests per 60-second window
- **Rate Limiting**: Dual-layer global and per-IP rate limiting (configurable via environment variables)
- **Content Types**: JSON for API, multipart/form-data for uploads
### Performance Characteristics
- **Throughput**: Designed for high-volume document processing
- **Latency**: Sub-second API response times for most operations
- **Availability**: Distributed architecture supports high availability
- **Durability**: S3 and RDS provide 99.999999999% (11 9's) and 99.95% durability respectively
- **Durability**: S3 and RDS provide 99.999999999% (11 9's) and 99.95% durability respectively
+44 -126
View File
@@ -2,106 +2,60 @@
## Service Catalog
### Primary Services
### Active Services
#### queryAPI - Main HTTP API Service
- **Purpose**: Central management API for all user entities and operations
- **Port**: 8080
- **Type**: HTTP REST API
- **Key Responsibilities**:
- Client management and authentication
- Document upload and metadata management (single and batch)
- Batch upload processing with ZIP archive support
- Folder and label management
- Field extraction configuration
- Export operations and status monitoring
- Background task coordination and monitoring
- User interface and API documentation
- **Responsibilities**:
- Client, document, folder, label, collector, export, field extraction, admin, and EULA endpoints
- Authentication/authorization middleware integration (Cognito + Permit.io)
- Batch upload background processing coordination
#### storeEventRunner - S3 Event Processor
- **Purpose**: Entry point for document processing pipeline
- **Port**: 8081
- **Type**: Event-driven queue consumer
- **Input**: S3 PUT event notifications (single files and batch uploads)
- **Output**: Events to DOCINIT queue
- **Responsibilities**: Processes S3 upload events, handles ZIP archive extraction, initiates document processing
- **Type**: Queue consumer
- **Input**: S3 event notification messages
- **Output**: DOCINIT queue messages
#### docInitRunner - Document Initialization
- **Purpose**: Document registration and duplicate detection
- **Port**: 8082
- **Type**: Queue-based processor
- **Input**: `{bucket, key, hash}` from DOCINIT queue
- **Output**: Events to DOCSYNC queue
- **Key Functions**:
- ETag-based duplicate detection
- Document reference entity creation
- Upload success logging
- **Type**: Queue consumer
- **Input**: DOCINIT queue messages
- **Output**: DOCSYNC queue messages
- **Responsibilities**: document registration and deduplication by client/hash
#### docSyncRunner - Document Sync Validation
- **Purpose**: Validates document eligibility for processing
#### docSyncRunner - Document Sync Gate
- **Port**: 8083
- **Type**: Queue-based processor
- **Input**: `{document_id}` from DOCSYNC queue
- **Output**: Events to DOCCLEAN queue
- **Logic**: Validates client `can_sync` parameter
- **Type**: Queue consumer
- **Input**: DOCSYNC queue messages
- **Output**: DOCCLEAN queue messages
- **Responsibilities**: enforce client sync eligibility before processing
#### docCleanRunner - Document Preparation
- **Purpose**: Document cleaning and validation
#### docCleanRunner - Document Validation/Cleaning
- **Port**: 8084
- **Type**: Queue-based processor
- **Input**: `{document_id}` from DOCCLEAN queue
- **Output**: Events to DOCTEXT queue
- **Dependencies**: S3 ObjectStore for document access
- **Type**: Queue consumer
- **Input**: DOCCLEAN queue messages
- **Output**: none (pipeline currently ends here)
- **Responsibilities**: content validation, clean record creation, idempotent processing
#### docTextRunner - Text Extraction
- **Purpose**: OCR and text extraction using AWS Textract
- **Port**: 8085
- **Type**: Queue-based processor
- **Input**: `{document_id}` from DOCTEXT queue
- **Output**: Text stored in database, processing complete
- **Dependencies**: AWS Textract, S3 ObjectStore
### Synchronization Services
#### clientSyncRunner - Client-Level Synchronization
- **Purpose**: Synchronizes all documents for a client
#### clientSyncRunner - Client Reprocessing Trigger
- **Port**: 8089
- **Type**: Queue-based processor
- **Input**: `{client_id}` from CLIENTSYNC queue
- **Output**: Events to DOCSYNC queue for each client document
- **Type**: Queue consumer
- **Input**: CLIENTSYNC queue messages
- **Output**: DOCSYNC queue messages (one per client document)
### Utility Services
### Deprecated Compatibility Services
#### healthcheck - Health Monitoring
- **Purpose**: Service health validation
- **Type**: HTTP health check utility
- **Target**: localhost:8080 health endpoint
#### textExtractor - Development Utility
- **Purpose**: Local text extraction for testing
- **Type**: Standalone utility
- **Use Case**: Development and testing workflows
#### metricsExample_test - Monitoring Example
- **Purpose**: Demonstrates comprehensive Prometheus metrics implementation
- **Type**: Educational/testing service
- **Features**: Showcases 12 standard metric types (Counter, Gauge, Histogram, Summary)
#### Background Task Framework
- **Purpose**: Comprehensive background task execution system
- **Type**: Embedded framework within services
- **Features**:
- Periodic task execution with configurable intervals
- On-demand task triggering via signals
- Work coalescing to prevent duplicate processing
- Statistics tracking and performance monitoring
- Graceful shutdown with task completion
The following services are retained as deployment compatibility placeholders and currently run health-only behavior:
- `docTextRunner` (8085)
- `querySyncRunner` (8087)
- `queryRunner` (8088)
- `queryVersionSyncRunner` (8090)
## Data Flow Architecture
### Document Processing Pipeline
[link to rendered version here](https://mermaid.live/edit#pako:eNp9k8FPgzAUxv-VhoMnPe3GYckGmCwxKLCpS7zU8rY1QottMVuW_e_2lWKGiBzIS78f3-v7Ss8BkyUEIQk0fLYgGMSc7hWt3wSxT0OV4Yw3VBgSVRyEGa8Xsz_WjFRAqCYai-TLfpi3QoAaoyvBDZKlZFhOYcVJMI9hOYVFFVDhOVdPgWs49m2xnMKyvq8NR53-65wh8ENOUfESkZga-k41dHr37uK9m8-LWUg2TSVpSWLJ2von9GKGKgYa2pq4VEkqDd9xRg2XwmNIWBLTDEn8GK3S1Rq31_p-KFg9XoYkOgD7IHHbVGgBegDgtM6g2KbRtQEK3uCZVry0X_rtO2kAuUNwLtFDskivbZzkB-4nJU9KMtCai_0QwkNyNuvkdTANCt4kh1FmXl28FKGrFWXjHh7KuoGzTZJvf4-cXc18z0VJkiOw1p5iBe7keZ9dD7rfwZsNjHDdG3UeQG78fclBt5XRwS0JalA15aW9mOfAHKB2V7SEHbVAcLl8AzPQPDE=)
### Active Document Processing Pipeline
```mermaid
sequenceDiagram
@@ -111,21 +65,12 @@ sequenceDiagram
participant Init as docInitRunner
participant Sync as docSyncRunner
participant Clean as docCleanRunner
participant Text as docTextRunner
participant DB as Database
Client->>S3: Upload Document
S3->>Store: S3 Event Notification
Store->>Init: DOCINIT Queue
Init->>DB: Check Duplicates
Init->>Sync: DOCSYNC Queue
Sync->>DB: Validate Client Sync
Sync->>Clean: DOCCLEAN Queue
Clean->>S3: Document Processing
Clean->>Text: DOCTEXT Queue
Text->>S3: Read Document
Text->>AWS: Textract Processing
Text->>DB: Store Extracted Text
```
### Client Synchronization Flow
@@ -135,59 +80,32 @@ sequenceDiagram
participant Admin
participant API as queryAPI
participant CS as clientSyncRunner
participant Pipeline as Document Pipeline
participant Sync as docSyncRunner
Admin->>API: Update Client Configuration
Admin->>API: Trigger client sync
API->>CS: CLIENTSYNC Queue
CS->>Pipeline: DOCSYNC Queue (per client document)
Pipeline->>Pipeline: Reprocess Documents
CS->>Sync: DOCSYNC Queue (per client document)
```
## Communication Patterns
### Inter-Service Messaging
All services communicate through **AWS SQS queues** with the following patterns:
- **Fire-and-forget**: Services publish events without waiting for responses
- **Queue-based**: Loose coupling between services
- **Retry logic**: Built-in message retry and dead letter queue support
- **Ordering**: FIFO queues where order matters
- Services use **AWS SQS** for asynchronous decoupling
- Queue consumers are idempotent and retry-friendly
### Queue Architecture
### Queue Architecture (Active)
| Queue Name | Producer | Consumer | Message Format |
|------------|----------|----------|----------------|
| DOCINIT | storeEventRunner | docInitRunner | `{bucket, key, hash}` |
| DOCSYNC | docInitRunner, clientSyncRunner | docSyncRunner | `{document_id}` |
| DOCCLEAN | docSyncRunner | docCleanRunner | `{document_id}` |
| DOCTEXT | docCleanRunner | docTextRunner | `{document_id}` |
| CLIENTSYNC | queryAPI | clientSyncRunner | `{client_id}` |
| BATCH_PROCESSING | queryAPI (batch worker) | Internal batch processor | `{batch_id, files[]}` |
### Database Access Patterns
- **Repository Pattern**: SQLC-generated type-safe database access
- **Connection Pooling**: Shared database connections across services
- **Transaction Management**: ACID transactions for data consistency
- **Read/Write Separation**: Potential for read replica scaling
### Queue Architecture (Deprecated Compatibility)
## Scalability and Performance
| Queue Name | Notes |
|------------|-------|
| DOCTEXT | Kept in some deployments for compatibility; not part of active pipeline |
| QUERYSYNC / QUERY / QUERYVERSIONSYNC | Legacy query pipeline queues; query subsystem removed |
### Horizontal Scaling Strategy
- **API Service**: Load balancer with multiple instances
- **Queue Processors**: Auto-scaling based on queue depth
- **Database**: Read replicas for scaling
- **Storage**: S3 scales automatically
### Performance Optimizations
- **Connection Pooling**: Database connection reuse
- **Batch Processing**: Queue messages processed in batches
- **Caching**: JWKS caching for authentication
- **Compression**: Gzip compression for API responses
### Monitoring and Observability
- **Metrics**: Comprehensive Prometheus metrics with 12 standard types (Counter, Gauge, Histogram, Summary)
- Request metrics, performance tracking, resource usage, dependency health
- Cache operations, background tasks, queue metrics, error tracking
- **Tracing**: OpenTelemetry distributed tracing with correlation IDs
- **Logging**: Structured logging with comprehensive correlation
- **Health Checks**: Built-in health endpoints with dependency validation
- **Background Task Monitoring**: Task execution statistics and performance tracking
+3 -3
View File
@@ -70,14 +70,14 @@ OAuth2 callback handler that processes authentication response.
Terminates user session and clears authentication cookies.
**Security**: Requires JWT authentication
**Security**: Public (no authentication required)
**Response**: `302` Redirect to Cognito logout page or application home
#### `GET /home`
Returns the HTML menu page for authenticated users.
Returns the HTML menu page.
**Security**: Requires JWT authentication
**Security**: Public (no authentication required)
**Response**: `200` HTML content
#### `GET /identity`
+11 -779
View File
@@ -2,7 +2,9 @@
## Database Schema Overview
The system uses PostgreSQL 17.2 with a sophisticated schema designed for multi-tenant document processing, query orchestration, and comprehensive versioning. The schema is managed through database migrations and uses SQLC for type-safe Go integration.
The system uses PostgreSQL 17.2 with a schema designed for multi-tenant document processing, field extraction, and versioning. The schema is managed through migrations and uses SQLC for type-safe Go integration.
> Note: Query/result tables and text-extraction tables were removed in migrations `00000000000119` and `00000000000120`. Any references to those tables in older sections should be treated as legacy context, not active runtime architecture.
## Entity Relationship Diagram
@@ -1055,790 +1057,20 @@ CREATE TABLE documentTextExtractionEntries (
- Enables version-based filtering of text extraction results
- Used by `currentTextEntries` view to get latest valid extractions
### Processing Dependencies
[link to rendered version here](https://mermaid.live/edit#pako:eNo9jz0KwzAMRq9iNDcX6NClZOsU2inNYGy1Mfgn2AokhNy9tkytRTx9TyAdoIJGuAr4RrnM4jG8vcj1WmyQetRBrQ49VUyT6Lqb6D3FvUWFDKap7nHG1t2i9M1i-ksMLD1xo-YU6DeKUpEJTS5TdgdMq6U0xtonuAhwGJ00Ot9_AM3o-BONH5kNOM8fK3VMWw==)
### Processing Dependencies (Current)
```mermaid
graph LR
Upload[documentUploads] --> Entry[documentEntries]
Entry --> Clean[documentCleans]
Clean --> Text[documentTextExtractions]
Text --> Results[results]
Clean --> FieldExtraction[documentFieldExtractions]
```
## Query Dependency System
## Legacy Sections Removed
### Dependency Management
```sql
-- Query-to-query dependencies
CREATE TABLE requiredQueries (
queryId uuid NOT NULL REFERENCES queries(queryId),
requiredQueryId uuid NOT NULL REFERENCES queries(queryId),
versionId int4 NOT NULL,
addedVersion int4 NOT NULL,
removedVersion int4,
CHECK (queryId != requiredQueryId)
);
The previous query subsystem and text extraction subsystem were removed from active runtime architecture. Legacy query/text schema details have been intentionally omitted from this generated doc to avoid drift.
-- Result-to-result dependencies
CREATE TABLE resultDependencies (
resultId uuid NOT NULL REFERENCES results(id),
requiredResultId uuid NOT NULL REFERENCES results(id),
CHECK (resultId != requiredResultId)
);
```
### Dependency Resolution Views
```sql
-- Recursive view for transitive dependencies
CREATE VIEW queryActiveDependencies AS
WITH RECURSIVE deps AS (
-- Direct dependencies
SELECT DISTINCT queryId, requiredQueryId, 1 as depth
FROM requiredQueries rq
JOIN queryActiveVersions qav ON rq.queryId = qav.queryId AND rq.versionId = qav.versionId
WHERE rq.removedVersion IS NULL
UNION
-- Transitive dependencies
SELECT d.queryId, rq.requiredQueryId, d.depth + 1
FROM deps d
JOIN requiredQueries rq ON d.requiredQueryId = rq.queryId
JOIN queryActiveVersions qav ON rq.queryId = qav.queryId AND rq.versionId = qav.versionId
WHERE rq.removedVersion IS NULL AND d.depth < 10 -- Cycle prevention
)
SELECT * FROM deps;
```
## Collector Configuration System
### Client-Query Mapping
```sql
-- Maps named queries to specific query implementations for clients
CREATE TABLE collectorQueries (
id uuid PRIMARY KEY DEFAULT uuid_generate_v7(),
clientId varchar(255) NOT NULL,
name varchar(255) NOT NULL,
queryId uuid NOT NULL,
addedVersion int NOT NULL,
removedVersion int,
FOREIGN KEY (queryId) REFERENCES queries(queryId),
FOREIGN KEY (clientId) REFERENCES clients(clientId),
FOREIGN KEY (clientId, addedVersion) REFERENCES collectorVersions(clientId, id),
FOREIGN KEY (clientId, removedVersion) REFERENCES collectorVersions(clientId, id),
UNIQUE (clientId, name, removedVersion)
);
```
**Key Features**:
- Maps human-readable query names to query implementations per client
- Version tracking with addedVersion/removedVersion
- Unique constraint on (clientId, name, removedVersion) allows name reuse across versions
### Version Thresholds
```sql
-- Minimum clean processing version required
CREATE TABLE collectorMinCleanVersions (
id uuid PRIMARY KEY DEFAULT uuid_generate_v7(),
clientId varchar(255) NOT NULL,
versionId bigint NOT NULL,
addedVersion int NOT NULL,
removedVersion int,
FOREIGN KEY (clientId, addedVersion) REFERENCES collectorVersions(clientId, id),
FOREIGN KEY (clientId, removedVersion) REFERENCES collectorVersions(clientId, id),
FOREIGN KEY (clientId) REFERENCES clients(clientId)
);
-- Minimum text extraction version required
CREATE TABLE collectorMinTextVersions (
id uuid PRIMARY KEY DEFAULT uuid_generate_v7(),
clientId varchar(255) NOT NULL,
versionId bigint NOT NULL,
addedVersion int NOT NULL,
removedVersion int,
FOREIGN KEY (clientId, addedVersion) REFERENCES collectorVersions(clientId, id),
FOREIGN KEY (clientId, removedVersion) REFERENCES collectorVersions(clientId, id),
FOREIGN KEY (clientId) REFERENCES clients(clientId)
);
```
**Purpose**:
- Controls which document processing versions are valid for a client
- Allows clients to ignore outdated processing results
- Version-tracked to support temporal queries
### Complex Dependency Views
```sql
-- Builds complete query dependency tree per client
CREATE VIEW collectorQueryDependencyTree AS
SELECT DISTINCT
cq.clientId,
cq.name as queryName,
qad.queryId,
qad.requiredQueryId,
qad.depth
FROM collectorQueries cq
JOIN queryActiveDependencies qad ON cq.queryId = qad.queryId
WHERE cq.removedVersion IS NULL;
```
## Database Views
The system uses extensive views to simplify complex queries and encapsulate business logic. Views are organized by domain.
### Query Management Views
#### Query Current Active Versions (`queryCurrentActiveVersions`)
Resolves the current active version for each query, handling cases where no active version is set.
```sql
CREATE VIEW queryCurrentActiveVersions AS
SELECT DISTINCT
q.queryId,
COALESCE(
(FIRST_VALUE(av.versionId) OVER (PARTITION BY q.queryId ORDER BY av.activeVersionEntryId DESC)),
0
)::int AS activeVersion
FROM queries AS q
LEFT JOIN queryActiveVersions AS av ON av.queryId = q.queryId;
```
**Purpose**: Provides latest active version per query, defaulting to 0 if none set.
#### Query Latest Versions (`queryLatestVersions`)
Tracks the highest version number for each query.
```sql
CREATE VIEW queryLatestVersions AS
SELECT
q.queryId,
COALESCE(MAX(v.versionId), 0)::int AS latestVersion
FROM queries AS q
LEFT JOIN queryVersions AS v ON v.queryId = q.queryId
GROUP BY q.queryId;
```
#### Query Current Configs (`queryCurrentConfigs`)
Returns the active configuration for each query based on current version.
```sql
CREATE VIEW queryCurrentConfigs AS
SELECT av.queryId, c.config
FROM queryCurrentActiveVersions AS av
LEFT JOIN queryConfigs AS c ON av.queryId = c.queryId
AND isInVersion(av.activeVersion, c.addedVersion, c.removedVersion);
```
#### Query Current Required IDs (`queryCurrentRequiredIds`)
Lists direct dependencies for each query at their current active version.
```sql
CREATE VIEW queryCurrentRequiredIds AS
SELECT DISTINCT av.queryId, r.requiredQueryId
FROM queryCurrentActiveVersions AS av
LEFT JOIN requiredQueries AS r ON av.queryId = r.queryId
AND isInVersion(av.activeVersion, r.addedVersion, r.removedVersion);
```
#### Query Current Required IDs Aggregated (`queryCurrentRequiredIdsAGG`)
Aggregates required query IDs into arrays for easier consumption.
```sql
CREATE VIEW queryCurrentRequiredIdsAGG AS
SELECT queryId,
COALESCE(
ARRAY_AGG(DISTINCT requiredQueryId)
FILTER (WHERE requiredQueryId != '00000000-0000-0000-0000-000000000000')::uuid[],
ARRAY[]::uuid[]
)::uuid[] AS requiredIds
FROM queryCurrentRequiredIds
GROUP BY queryId;
```
#### Full Active Queries (`fullActiveQueries`)
Complete query information with versions, configs, and dependencies.
```sql
CREATE VIEW fullActiveQueries AS
SELECT DISTINCT
q.queryId,
q.queryType,
av.activeVersion,
lv.latestVersion,
c.config,
r.requiredIds
FROM queries AS q
JOIN queryCurrentActiveVersions AS av ON q.queryId = av.queryId
JOIN queryLatestVersions AS lv ON lv.queryId = q.queryId
JOIN queryCurrentConfigs AS c ON q.queryId = c.queryId
JOIN queryCurrentRequiredIdsAGG AS r ON q.queryId = r.queryId;
```
**Purpose**: Single view providing all query metadata needed for execution.
#### Query Active Dependencies (`queryActiveDependencies`)
Recursive view computing transitive query dependencies with cycle detection.
```sql
CREATE VIEW queryActiveDependencies AS
WITH RECURSIVE queryActiveDependencies(queryId, requiredQueryId, path, cycle) AS (
SELECT
queryId,
requiredQueryId,
ARRAY[queryId, requiredQueryId]::uuid[] AS path,
false AS cycle
FROM queryCurrentRequiredIds
UNION ALL
SELECT
q.queryId,
qd.requiredQueryId,
path || qd.requiredQueryId,
qd.requiredQueryId = ANY(path) AS cycle
FROM queryCurrentRequiredIds AS q
JOIN queryActiveDependencies AS qd ON q.queryId = qd.requiredQueryId
WHERE NOT qd.cycle
)
SELECT DISTINCT queryId AS id, requiredQueryId
FROM queryActiveDependencies
WHERE NOT cycle;
```
**Key Features**:
- Computes full dependency tree (not just direct dependencies)
- Detects and prevents infinite loops from circular dependencies
- Used for query execution ordering
### Collector Management Views
#### Collector Current Active Versions (`collectorCurrentActiveVersions`)
Resolves active collector version per client.
```sql
CREATE VIEW collectorCurrentActiveVersions AS
SELECT
c.clientId,
COALESCE(v.versionId, 0)::int AS activeVersion
FROM clients c
LEFT JOIN (
SELECT DISTINCT ON (clientId)
clientId,
versionId
FROM collectorActiveVersions
ORDER BY clientId, id DESC
) v ON c.clientId = v.clientId;
```
#### Collector Latest Versions (`collectorLatestVersions`)
Tracks highest version number per collector.
```sql
CREATE VIEW collectorLatestVersions AS
SELECT
j.clientId,
COALESCE(MAX(v.id), 0)::int AS latestVersion
FROM clients AS j
LEFT JOIN collectorVersions AS v ON v.clientId = j.clientId
GROUP BY j.clientId;
```
#### Current Collector Min Text/Clean Versions
Views tracking minimum processing versions required per client.
```sql
CREATE VIEW currentCollectorMinTextVersions AS
SELECT DISTINCT
av.clientId,
COALESCE(ctv.versionId, 0) AS minTextVersion
FROM collectorCurrentActiveVersions AS av
LEFT JOIN collectorMinTextVersions AS ctv ON av.clientId = ctv.clientId
AND isInVersion(av.activeVersion, ctv.addedVersion, ctv.removedVersion);
CREATE VIEW currentCollectorMinCleanVersions AS
SELECT DISTINCT
av.clientId,
COALESCE(ctv.versionId, 0) AS minCleanVersion
FROM collectorCurrentActiveVersions AS av
LEFT JOIN collectorMinCleanVersions AS ctv ON av.clientId = ctv.clientId
AND isInVersion(av.activeVersion, ctv.addedVersion, ctv.removedVersion);
```
#### Current Collector Queries (`currentCollectorQueries`)
Active query mappings per client.
```sql
CREATE VIEW currentCollectorQueries AS
SELECT DISTINCT
av.clientId,
q.name,
q.queryId
FROM collectorCurrentActiveVersions AS av
LEFT JOIN collectorQueries AS q ON av.clientId = q.clientId
AND isInVersion(av.activeVersion, q.addedVersion, q.removedVersion);
```
#### Full Active Collectors (`fullActiveCollectors`)
Complete collector configuration with all metadata.
```sql
CREATE VIEW fullActiveCollectors AS
SELECT DISTINCT
av.clientId,
ccv.minCleanVersion,
ctv.minTextVersion,
av.activeVersion,
lv.latestVersion,
q.fields
FROM collectorCurrentActiveVersions AS av
JOIN collectorLatestVersions AS lv ON lv.clientId = av.clientId
JOIN currentCollectorMinCleanVersions AS ccv ON av.clientId = ccv.clientId
JOIN currentCollectorMinTextVersions AS ctv ON av.clientId = ctv.clientId
JOIN currentCollectorQueriesJSONAGG AS q ON av.clientId = q.clientId;
```
#### Collector Query Dependency Tree (`collectorQueryDependencyTree`)
Recursive view computing all queries (direct and transitive) needed by a collector.
```sql
CREATE VIEW collectorQueryDependencyTree AS
WITH RECURSIVE collectorQueryDependencyTree AS (
SELECT cq.clientId, cq.queryId, ri.requiredIds
FROM currentCollectorQueries AS cq
JOIN queryCurrentRequiredIdsAGG AS ri ON cq.queryId = ri.queryId
UNION ALL
SELECT acq.clientId, q.queryId, q.requiredIds
FROM queryCurrentRequiredIdsAGG AS q
JOIN collectorQueryDependencyTree AS acq ON q.queryId = ANY(acq.requiredIds)
)
SELECT DISTINCT
ct.clientId,
ct.queryId,
q.queryType,
av.activeVersion AS queryVersion,
ct.requiredIds
FROM collectorQueryDependencyTree AS ct
JOIN queryCurrentActiveVersions AS av ON ct.queryId = av.queryId
JOIN queries AS q ON q.queryId = ct.queryId;
```
**Purpose**: Used to determine all queries that must be executed for a client's configuration.
### Document Processing Views
#### Current Clean Entries (`currentCleanEntries`)
Latest valid clean entry per document meeting minimum version requirements.
```sql
CREATE VIEW currentCleanEntries AS
WITH RankedExtractions AS (
SELECT
dc.id, dc.documentId, dc.bucket, dc.key, dc.mimetype,
dc.fail, dc.hash, dce.version, d.clientId,
ROW_NUMBER() OVER (PARTITION BY dc.documentId ORDER BY dc.id DESC, dce.id DESC) AS row_num
FROM documents d
JOIN currentCollectorMinCleanVersions ccv ON d.clientId = ccv.clientId
JOIN documentCleans dc ON dc.documentId = d.id
JOIN documentCleanEntries dce ON dc.id = dce.cleanId
AND dce.version >= ccv.minCleanVersion
)
SELECT id, documentId, bucket, key, version, hash, mimetype, fail, clientId
FROM RankedExtractions
WHERE row_num = 1;
```
**Key Features**:
- Filters by collector's minimum clean version requirement
- Returns most recent clean entry per document
- Includes both successful and failed cleans
#### Current Text Entries (`currentTextEntries`)
Latest valid text extraction per document meeting minimum version requirements.
```sql
CREATE VIEW currentTextEntries AS
WITH RankedExtractions AS (
SELECT
extract.id, cc.documentId, extract.bucket, extract.key,
extract.hash, extract.cleanId, dtee.version AS extractionVersion,
ROW_NUMBER() OVER (PARTITION BY cc.documentId ORDER BY dtee.id DESC) AS row_num
FROM documents d
JOIN currentCleanEntries cc ON cc.documentId = d.id
JOIN currentCollectorMinTextVersions ctv ON ctv.clientId = d.clientId
JOIN documentTextExtractions extract ON extract.cleanId = cc.id
JOIN documentTextExtractionEntries dtee ON dtee.textId = extract.id
AND dtee.version >= ctv.minTextVersion
)
SELECT id, documentId, bucket, key, hash, cleanId, extractionVersion
FROM RankedExtractions
WHERE row_num = 1;
```
**Purpose**: Used to determine which text extractions are valid for query processing.
#### Current Client Can Sync (`currentClientCanSync`)
Latest sync permission status per client.
```sql
CREATE VIEW currentClientCanSync AS
WITH RankedExtractions AS (
SELECT
ccs.clientId,
ccs.canSync,
ROW_NUMBER() OVER (PARTITION BY ccs.clientId ORDER BY ccs.syncId DESC) AS row_num
FROM clients c
JOIN clientCanSync ccs ON c.clientId = ccs.clientId
)
SELECT
c.clientId,
COALESCE(r.canSync, false) AS canSync
FROM clients c
LEFT JOIN RankedExtractions r ON r.clientId = c.clientId AND r.row_num = 1;
```
#### Full Clients (`fullClients`)
Complete client information with sync status.
```sql
CREATE VIEW fullClients AS
SELECT c.clientId, c.name, cs.canSync
FROM clients AS c
JOIN currentClientCanSync AS cs ON cs.clientId = c.clientId;
```
## Advanced Database Features
### Business Logic Functions
#### Document Processing Functions
**listDocumentIDs**: Paginated document listing with total count.
```sql
CREATE OR REPLACE FUNCTION listDocumentIDs(
_clientId varchar(255),
_batchSize INTEGER,
_offset INTEGER
)
RETURNS TABLE (id uuid, totalCount BIGINT) AS $$
DECLARE
totalCount BIGINT := 0;
BEGIN
SELECT COUNT(*)::BIGINT INTO totalCount FROM documents WHERE clientId = _clientId;
RETURN QUERY
SELECT d.id, totalCount
FROM documents AS d
WHERE d.clientId = _clientId
ORDER BY d.id
LIMIT _batchSize
OFFSET _offset;
END;
$$ LANGUAGE plpgsql;
```
**Key Features**:
- Returns both document IDs and total count in single query
- Supports pagination via batch size and offset
- Orders by document ID for consistent paging
**collectorQueryDependencyTreeByClient**: Optimized function for getting query dependencies per client.
```sql
CREATE OR REPLACE FUNCTION collectorQueryDependencyTreeByClient(
_clientId varchar(255)
)
RETURNS TABLE (
clientId varchar(255),
queryId uuid,
queryType queryType,
queryVersion int,
requiredIds uuid[]
) AS $$
BEGIN
RETURN QUERY
WITH clientQueries AS (
SELECT q.clientId, q.queryId
FROM currentCollectorQueries AS q
WHERE q.clientId = _clientId
),
dependencyTree AS (
WITH RECURSIVE collectorQueryDependencyTree AS (
SELECT cq.clientId, cq.queryId, ri.requiredIds
FROM clientQueries AS cq
JOIN queryCurrentRequiredIdsAGG AS ri ON cq.queryId = ri.queryId
UNION ALL
SELECT acq.clientId, q.queryId, q.requiredIds
FROM queryCurrentRequiredIdsAGG AS q
JOIN collectorQueryDependencyTree AS acq ON q.queryId = ANY(acq.requiredIds)
)
SELECT DISTINCT
ct.clientId, ct.queryId, q.queryType,
av.activeVersion AS queryVersion, ct.requiredIds
FROM collectorQueryDependencyTree AS ct
JOIN queryCurrentActiveVersions AS av ON ct.queryId = av.queryId
JOIN queries AS q ON q.queryId = ct.queryId
)
SELECT t.clientId, t.queryId, t.queryType, t.queryVersion, t.requiredIds
FROM dependencyTree AS t
WHERE t.clientID IS NOT NULL;
END;
$$ LANGUAGE plpgsql;
```
**Purpose**: More efficient than view for single-client queries.
#### Result Validation Functions
**listValidDocumentResults**: Complex dependency resolution for query results.
```sql
CREATE OR REPLACE FUNCTION listValidDocumentResults(doc_id uuid)
RETURNS TABLE (documentId uuid, queryId uuid, resultId uuid, value text) AS $$
DECLARE
count_before INT;
count_after INT;
iterations INT := 0;
BEGIN
-- Creates temporary tables and iteratively resolves dependencies
-- Full implementation in migration 00000000000102_document_views.up.sql:103-211
END;
$$ LANGUAGE plpgsql;
```
**Key Features**:
- Validates that all query dependencies have been satisfied
- Uses temporary tables for efficient processing
- Iteratively builds valid result set
- Checks both query dependencies and result dependencies
- Returns only results where all prerequisites are met
### Database Triggers
The schema uses several triggers to automate versioning and configuration management:
#### Query Version Auto-Increment
```sql
CREATE OR REPLACE FUNCTION setQueryVersionNumber()
RETURNS TRIGGER AS $$
BEGIN
SELECT COALESCE(MAX(versionId), 0) + 1
INTO NEW.versionId
FROM queryVersions
WHERE queryId = NEW.queryId;
RETURN NEW;
END;
$$ LANGUAGE plpgsql;
CREATE TRIGGER setQueryVersionNumberTrigger
BEFORE INSERT ON queryVersions
FOR EACH ROW
EXECUTE FUNCTION setQueryVersionNumber();
```
#### Collector Version Auto-Increment
```sql
CREATE OR REPLACE FUNCTION setCollectorVersionNumber()
RETURNS TRIGGER AS $$
BEGIN
SELECT COALESCE(MAX(id), 0) + 1
INTO NEW.id
FROM collectorVersions
WHERE clientId = NEW.clientId;
RETURN NEW;
END;
$$ LANGUAGE plpgsql;
CREATE TRIGGER setCollectorVersionNumberTrigger
BEFORE INSERT ON collectorVersions
FOR EACH ROW
EXECUTE FUNCTION setCollectorVersionNumber();
```
#### Query Config Version Management
```sql
CREATE OR REPLACE FUNCTION removeQueryConfig()
RETURNS TRIGGER AS $$
BEGIN
UPDATE queryConfigs
SET removedVersion = NEW.addedVersion
WHERE queryId = NEW.queryId AND removedVersion IS NULL;
RETURN NEW;
END;
$$ LANGUAGE plpgsql;
CREATE TRIGGER removeQueryConfigTrigger
BEFORE INSERT ON queryConfigs
FOR EACH ROW
EXECUTE FUNCTION removeQueryConfig();
```
**Purpose**: Automatically closes out previous configs when new ones are added.
#### Collector Min Version Management
Similar triggers exist for `collectorMinCleanVersions` and `collectorMinTextVersions` to automatically set `removedVersion` when new minimum versions are added.
## Data Integrity and Constraints
### Referential Integrity
- All foreign key relationships enforced at database level
- No cascade deletes (explicit deletion required for safety)
- Unique constraints prevent duplicate business entities
- Composite foreign keys ensure version referential integrity
### Data Validation Constraints
#### Enum Types
- `queryType`: `'context_full'`, `'json_extractor'`
- `cleanMimeType`: `'application/pdf'`
- `cleanFailType`: Various failure reasons (invalid_mimetype, invalid_read, etc.)
- `batch_status`: `'processing'`, `'completed'`, `'failed'`, `'cancelled'`
#### Check Constraints
- `requiredQueries`: Self-reference prevention (`queryId != requiredQueryId`)
- `resultDependencies`: Self-reference prevention (`resultId != requiredResultId`)
- `documentCleans`: XOR constraint ensures either success OR failure, not both
- `bucket_and_key_together`: Bucket, key, mimetype, hash must all be present or all null
- `location_xor_fail`: Cannot have both location (success) and fail (failure) indicators
#### Custom Domain Types
- `unsignedsmallint`: SMALLINT constrained to non-negative values
- Used for part numbers in storage distribution
### Audit and Traceability
- UUID v7 generation provides time-ordered insertion
- Version tracking enables complete audit trails
- Temporal validity functions support point-in-time queries
- Triggers automatically manage version transitions
- All critical tables have UUID primary keys for global uniqueness
## Scalability Considerations
### Indexing Strategy
- Primary keys automatically indexed (UUID v7)
- Foreign keys automatically indexed
- Composite indexes on frequent query patterns
- Unique constraints create secondary indexes
### Partitioning Opportunities
- Time-based partitioning using UUID v7 timestamp component
- Client-based partitioning for multi-tenant isolation
- Archive strategies for old versions
### Performance Optimizations
- Window functions for latest version queries
- Recursive CTEs with depth limits for dependency resolution
- Pagination support in all list functions
- Efficient temporal queries using `isInVersion()` function
## Migration Strategy
### Database Evolution
- Sequential numbered migrations for version control (00000000000001, 00000000000002, etc.)
- Separate UP/DOWN migrations for rollback capability
- Extension management for UUID generation and other features
- View-based abstractions for query flexibility
- Migration files located in: `internal/database/migrations/`
### Migration Numbering Convention
- `00000000000001-00000000000099`: Core tables (extensions, queries, clients, collectors, documents, results)
- `00000000000100-00000000000199`: Views and functions
- `00000000000200+`: Schema modifications and new features
### Recent Schema Changes
- **Migration 103**: Added `batch_uploads` table and `documents.batch_id`
- **Migration 104**: Added S3 storage columns to `batch_uploads` (`archive_bucket`, `archive_key`, `file_size_bytes`) with storage consistency constraint
- **Migration 105**: Added `documents.filename` column
- **Migration 106**: Added `documentUploads.filename` column
- **Migration 107**: Added `documentUploads.batch_id` column (without foreign key constraint)
- **Migration 108**: Added index on `documentUploads.clientId`
- **Migration 109**: Added `folders` table for virtual folder hierarchy with self-referential parentId
- **Migration 110**: Added `labels` and `documentLabels` tables for document processing status tracking
- **Migration 111**: Added `documents.folderId` and `documents.originalPath` columns; added comments for S3 immutability
- **Migration 112**: Added `documentFieldExtractions` table with 19 single-value fields
- **Migration 113**: Added `documentFieldExtractionVersions` table for version tracking
- **Migration 114**: Added `documentFieldExtractionArrayFields` table with 112 array fields
- **Migration 115**: Added `currentFieldExtractions` and `currentFieldExtractionsWithArrayCount` views
- **Migration 116**: Added `documentUploads.folder_id` column with foreign key constraint
- **Migration 121**: Added `eulaVersions` and `eulaAgreements` tables for EULA management and consent tracking
### Version Compatibility
- Backward-compatible schema changes prioritized
- Graceful handling of version mismatches via views
- New columns added as nullable to avoid breaking existing code
- Indexes added separately from table creation for safety
## Key Schema Patterns
### Naming Conventions
- **Tables**: Generally use camelCase (e.g., `clientId`, `documentId`)
- **Exception**: `batch_uploads` uses snake_case (`client_id`, `original_filename`)
- **Views**: camelCase naming (e.g., `currentCleanEntries`, `fullActiveQueries`)
- **Functions**: camelCase naming (e.g., `listDocumentIDs`, `isInVersion`)
### Common Data Types
- **IDs**: `uuid` with `DEFAULT uuid_generate_v7()` for time-ordered insertion
- **Client references**: `varchar(255)` for clientId
- **Text fields**: `TEXT` type (not varchar) for unlimited length
- **Timestamps**: `timestamp` without timezone
- **Versions**: `int` or `bigint` depending on expected range
- **Part numbers**: `unsignedsmallint` (0-32767)
- **JSON data**: `jsonb` for queryable JSON storage
### Versioning Pattern
Most versioned entities follow this pattern:
1. Core table with basic info (e.g., `queries`)
2. Versions table with composite PK (e.g., `queryVersions`)
3. Active version tracking table (e.g., `queryActiveVersions`)
4. Related data with `addedVersion`/`removedVersion` (e.g., `queryConfigs`)
5. Views to compute current state (e.g., `queryCurrentActiveVersions`)
6. Triggers to auto-increment versions
### View Pattern
Views are layered for reusability:
1. **Base views**: Compute current active versions
2. **Aggregation views**: Roll up related data (e.g., `queryCurrentRequiredIdsAGG`)
3. **Full views**: Join everything together (e.g., `fullActiveQueries`)
4. **Recursive views**: Compute transitive relationships (e.g., `queryActiveDependencies`)
### Performance Considerations
- Views use `DISTINCT` liberally to handle version overlap
- Window functions (`ROW_NUMBER`, `FIRST_VALUE`) for latest record selection
- Recursive CTEs include cycle detection
- Indexes created on foreign keys and frequently filtered columns
- Pagination functions return total count to avoid separate COUNT query
## Schema Statistics
### Table Count
- **Core entities**: 5 (clients, documents, batch_uploads, queries, results)
- **Organization tables**: 3 (folders, labels, documentLabels)
- **Field extraction tables**: 3 (documentFieldExtractions, documentFieldExtractionVersions, documentFieldExtractionArrayFields)
- **EULA tables**: 2 (eulaVersions, eulaAgreements)
- **Supporting tables**: 16 (versions, entries, configs, dependencies, etc.)
- **Total tables**: 29
### View Count
- **Query management**: 6 views
- **Collector management**: 7 views
- **Document processing**: 3 views
- **Field extraction**: 2 views (currentFieldExtractions, currentFieldExtractionsWithArrayCount)
- **Total views**: 18+
### Function Count
- **Business logic**: 3 main functions
- **Trigger functions**: 5 functions
- **Utility functions**: 2 (uuid_generate_v7, isInVersion)
- **Total functions**: 10+
### Enum Types
- `queryType`: 2 values
- `cleanMimeType`: 1 value
- `cleanFailType`: 9 values
- `batch_status`: 4 values
- **Total enums**: 4 types, 16 values
For removal details, see:
- `internal/database/migrations/00000000000119_remove_queries.up.sql`
- `internal/database/migrations/00000000000120_remove_text_extraction.up.sql`
+71 -333
View File
@@ -2,383 +2,121 @@
## Prerequisites
### Required Tools
- **Devbox**: Provides reproducible development environment with Go 1.24.3
- Ubuntu/MacOS: `curl -fsSL https://get.jetify.com/devbox | bash`
- NixOS: `devbox` (already available)
- **Docker**: Container runtime for services and testing
- Install from: https://docs.docker.com/engine/install/
- **Git**: Version control (usually pre-installed)
- **AWS CLI**: Required for Textract credentials and S3 access (installed via devbox)
- Devbox
- Docker
- Go 1.24.x (provided by devbox)
- Task (`task`)
### System Requirements
- **OS**: Linux, macOS, or Windows with WSL2
- **RAM**: 8GB minimum, 16GB recommended
- **Disk**: 10GB free space for development environment
- **Network**: Internet access for downloading dependencies
## Local Setup
### Docker Group Setup (Linux)
Ensure your user can run Docker commands without sudo:
```bash
sudo groupadd docker
sudo usermod -aG docker $USER
# Log out and back in for changes to take effect
```
## Local Development Setup
### 1. Environment Initialization
```bash
# Clone repository
git clone <repository-url>
cd query-orchestration.2
# Create environment file for custom variables
touch .env
# Enter development environment (installs all tools automatically)
devbox shell
```
### 2. Verify Installation
## Verify Tooling
```bash
# Check Go version
go version # Should show 1.24.0
# Check Docker
go version
docker --version
# Check Task runner
task --version # Should show 3.39.2+
# List available tasks
task --version
task --list
```
### 3. Full System Validation
Run the complete test suite to ensure everything works:
## Common Tasks
### Generate and Build
```bash
task generate
task build
```
### Full Test/Validation Suite
```bash
task fullsuite
```
This command will:
1. Generate latest API specifications
2. Run linting commands
3. Execute unit tests
4. Run end-to-end integration tests
5. Validate coverage thresholds (80% total, 60% function)
### Local Compose Lifecycle
## Development Workflow
### Common Development Tasks
#### Running Services Locally
```bash
# Start all services with local infrastructure
task dev:up
# Start specific service
task dev:api # Just the API service
task dev:runners # Just the queue runners
task compose:up
task compose:down
task compose:clean
```
#### API Development
### Linting
```bash
# Generate API code from OpenAPI spec
task openapi:generate
# Start API server with hot reload
task dev:api
# View API documentation
open http://localhost:8080/swagger/index.html
```
#### Database Operations
```bash
# Run database migrations
task db:migrate
# Reset database (development only)
task db:reset
# Generate database models from SQL
task db:generate
```
#### Testing
```bash
# Run unit tests
task test:unit
# Run integration tests
task test:integration
# Run tests with coverage
task test:coverage
# Run specific test package
go test ./internal/client/...
```
#### Code Quality
```bash
# Run linters
task lint
# Format code
task format
# Generate mocks for testing
task generate:mocks
```
### Development Environment Configuration
## Useful Focused Tasks
```bash
task openapi:generate
task db:generate
task db:lint
task test:functional
task test:race
```
## Key Local Ports
- `queryAPI`: `8080`
- `storeEventRunner`: `8081`
- `docInitRunner`: `8082`
- `docSyncRunner`: `8083`
- `docCleanRunner`: `8084`
- `clientSyncRunner`: `8089`
- `prometheus`: `9091`
- `postgres`: `5432`
- `localstack`: `4566`
- `permit_pdp`: `7766`
### Deprecated compatibility services (still present in compose in some environments)
- `docTextRunner`: `8085`
- `querySyncRunner`: `8087`
- `queryRunner`: `8088`
- `queryVersionSyncRunner`: `8090`
## Minimal `.env` Example
#### Environment Variables Hierarchy
1. **devbox.json "env"** section (highest priority)
2. **devbox.json "init_hook"** exports
3. **.env file** variables (lowest priority)
#### Key Development Variables
Create a `.env` file with:
```bash
# Database (optional - defaults provided)
PGUSER=postgres
PGPASSWORD=password
PGDATABASE=queryorchestration
PGHOST=localhost
PGPORT=5432
DB_NOSSL=true
# AWS (uses LocalStack in development)
AWS_REGION=us-east-1
AWS_ACCESS_KEY_ID=test
AWS_SECRET_ACCESS_KEY=test
AWS_SESSION_TOKEN=test
AWS_ENDPOINT_URL=http://localhost:4566
AWS_S3_USE_PATH_STYLE=true
# Authentication (disable for local development)
DISABLE_AUTH=true
# Logging
LOG_LEVEL=DEBUG
BUCKET=<your-bucket-name>
```
#### Service Ports (Local Development)
- **queryAPI**: 8080 (main API)
- **storeEventRunner**: 8081
- **docInitRunner**: 8082
- **docSyncRunner**: 8083
- **docCleanRunner**: 8084
- **docTextRunner**: 8085
- **querySyncRunner**: 8087
- **queryRunner**: 8088
- **clientSyncRunner**: 8089
- **queryVersionSyncRunner**: 8090
- **Prometheus**: 9091
- **PostgreSQL**: 5432
- **LocalStack**: 4566
## Troubleshooting
## Common Development Workflows
### 1. API Endpoint Development
```bash
# 1. Modify OpenAPI specification
vim serviceAPIs/queryAPI.yaml
# Validate compose files
task compose:lint
# 2. Regenerate API code
# Check migrations/codegen
task db:generate
task openapi:generate
# 3. Implement controller logic
vim api/queryAPI/controllers.go
# 4. Run API server
task dev:api
# 5. Test endpoint
curl -X GET http://localhost:8080/health
# Recreate local containers/volumes
task compose:clean
task compose:up
```
### 2. Database Schema Changes
```bash
# 1. Create new migration files
vim internal/database/migrations/000000000000X_feature.up.sql
vim internal/database/migrations/000000000000X_feature.down.sql
# 2. Add queries to SQL files
vim internal/database/queries/feature.sql
# 3. Regenerate database models
task db:generate
# 4. Run migrations
task db:migrate
# 5. Update Go service code
vim internal/feature/service.go
```
### 3. New Service Development
```bash
# 1. Implement controller interface
vim api/newRunner/runner.go
# 2. Create main command
vim cmd/newRunner/main.go
# 3. Add service configuration
vim internal/serviceconfig/newrunner/config.go
# 4. Update docker-compose
vim deployments/compose.local.yaml
# 5. Run tests
task test:integration
```
### 4. Adding New Queue
```bash
# 1. Add queue configuration
vim internal/serviceconfig/queue/newqueue/config.go
# 2. Update producer services
vim internal/service/producer.go
# 3. Update consumer services
vim api/consumerRunner/runner.go
# 4. Update environment variables
vim deployments/compose.local.yaml
```
## Testing Guidelines
### Unit Testing
- **Location**: `*_test.go` files alongside source code
- **Pattern**: Table-driven tests preferred
- **Mocking**: Use generated mocks for external dependencies
- **Coverage**: Minimum 60% function coverage required
Example unit test:
```go
func TestService_CreateClient(t *testing.T) {
tests := []struct {
name string
input CreateClientRequest
expected Client
wantErr bool
}{
{
name: "valid client creation",
input: CreateClientRequest{
ClientID: "test-client",
Name: "Test Client",
},
expected: Client{
ClientID: "test-client",
Name: "Test Client",
},
wantErr: false,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
// Test implementation
})
}
}
```
### Integration Testing
- **Location**: `test/` directory and `*_test.go` files
- **Infrastructure**: Uses testcontainers for real dependencies
- **Isolation**: Each test gets clean environment
- **Coverage**: Minimum 80% total coverage required
### End-to-End Testing
- **Environment**: Full service deployment via Docker Compose
- **Scope**: Complete workflows from API to database
- **Authentication**: Uses test Cognito setup or disabled auth
## Debugging and Troubleshooting
### Common Issues
#### Docker Permission Errors
```bash
# Fix Docker permissions (Linux)
sudo groupadd docker
sudo usermod -aG docker $USER
# Log out and log back in
```
#### Database Connection Issues
```bash
# Check PostgreSQL status
task db:status
# Reset database
task db:reset
# Check connection manually
psql -h localhost -p 5432 -U postgres -d queryorchestration
```
#### Service Startup Issues
```bash
# Check service logs
docker-compose -f deployments/compose.local.yaml logs [service-name]
# Check environment variables
task env:check
# Verify ports aren't in use
lsof -i :8080
```
#### Code Generation Issues
```bash
# Clean generated files
task clean
# Regenerate everything
task generate:all
# Check for syntax errors in source files
task lint
```
### Development Tools
#### Debugging
- **Delve**: Go debugger integrated with devbox
- **IDE Integration**: VS Code and GoLand support provided
- **Logging**: Structured logging with slog at DEBUG level
#### Monitoring
- **Prometheus**: Metrics at http://localhost:9091
- **Swagger UI**: API docs at http://localhost:8080/swagger/
- **Health Checks**: http://localhost:8080/health
#### Database Tools
- **psql**: Command-line PostgreSQL client
- **Database browser**: Connect to localhost:5432
- **Migration status**: `task db:status`
### Performance Profiling
```bash
# CPU profiling
go test -cpuprofile=cpu.prof -bench=.
# Memory profiling
go test -memprofile=mem.prof -bench=.
# Race detection
go test -race ./...
```
## Next Steps
After completing the setup:
1. **Explore the API**: Visit http://localhost:8080/swagger/
2. **Review Architecture**: Read the service architecture documentation
3. **Check Data Model**: Examine database schema documentation
4. **Run Examples**: Try the metrics example service
5. **Write Tests**: Create tests for new features
+4 -5
View File
@@ -18,7 +18,6 @@ query-orchestration.2/
├── docs/ # Documentation
├── scripts/ # Build and automation scripts
├── test/ # Integration and e2e tests
├── mocks/ # Generated mock implementations
├── assets/ # Test data and sample files
└── vendor/ # Vendored dependencies
```
@@ -56,8 +55,8 @@ Core business logic organized by domain boundaries:
```
internal/
├── client/ # Client management domain
├── document/ # Document processing domain
├── query/ # Query definition and execution
├── document/ # Document processing domain
├── fieldextraction/ # Field extraction domain
├── collector/ # Client-specific configurations
└── export/ # Data export operations
```
@@ -171,7 +170,7 @@ controller := api.NewController(svc)
## Naming Conventions
### Package Names
- **Domain packages**: Singular nouns (`client`, `document`, `query`)
- **Domain packages**: Singular nouns (`client`, `document`, `fieldextraction`)
- **Utility packages**: Descriptive (`serviceconfig`, `cognitoauth`)
- **Generated packages**: Match source (`queryAPI` from `queryAPI.yaml`)
@@ -397,4 +396,4 @@ func TestIntegration_DocumentProcessing(t *testing.T) {
- **Validation**: Automatic request/response validation
- **Documentation**: Auto-generated from specification
This organization enables maintainable, testable, and scalable code while following Go best practices and domain-driven design principles.
This organization enables maintainable, testable, and scalable code while following Go best practices and domain-driven design principles.
+113 -557
View File
@@ -1,560 +1,116 @@
# Configuration Management
## Updated Configuration (September 2025)
This document reflects the current configuration model implemented under `internal/serviceconfig`.
## Loading Model
Configuration is loaded through:
1. `.env` (if present)
2. Environment variables
3. Struct defaults (`envDefault`)
Runtime initialization is handled by `serviceconfig.InitializeConfig`.
## Core Configuration (`internal/serviceconfig/common.go`)
- `PORT` (default `8080`)
- `BASE_URL` (default `http://localhost`)
- Embedded config domains:
- auth
- thread pool
- logger
- observability
- database
- aws
- queue
## Database Configuration (`internal/serviceconfig/database/config.go`)
Required:
- `PGUSER`
- `PGPASSWORD`
- `PGHOST`
- `PGPORT`
- `PGDATABASE`
Optional:
- `DB_NOSSL` (default `false`)
## AWS/Object Store Configuration
### AWS (`internal/serviceconfig/aws/config.go`)
- `AWS_REGION` (required)
- `AWS_ACCESS_KEY_ID` / `AWS_SECRET_ACCESS_KEY` / `AWS_SESSION_TOKEN` (optional; SDK chain also supported)
- `AWS_ENDPOINT_URL` (optional global endpoint override)
- `AWS_PROFILE` (optional)
### Object Store (`internal/serviceconfig/objectstore/config.go`)
- `BUCKET` (required)
- `AWS_ENDPOINT_URL_S3` (optional S3-specific endpoint)
- `AWS_S3_USE_PATH_STYLE` (default `false`)
### Queue Client (`internal/serviceconfig/queue/config.go`)
- `AWS_ENDPOINT_URL_SQS` (optional SQS-specific endpoint)
## Queue URL Variables (service-specific)
Active flow:
- `STORE_EVENT_URL`
- `DOCUMENT_INIT_URL`
- `DOCUMENT_SYNC_URL`
- `DOCUMENT_CLEAN_URL`
- `CLIENT_SYNC_URL`
Compatibility/deprecated flow (still used by some deployment files):
- `DOCUMENT_TEXT_URL`
- `QUERY_SYNC_URL`
- `QUERY_URL`
- `QUERY_VERSION_SYNC_URL`
Per-runner listener queue URL:
- `QUEUE_URL` (required by runner base config)
## Authentication / Authorization
From `internal/serviceconfig/auth/config.go`:
- `COGNITO_REGION` (default `us-east-2`)
- `COGNITO_USER_POOL_ID` (required)
- `COGNITO_CLIENT_SECRET` (required)
- `COGNITO_DOMAIN` (required)
- `COGNITO_CLIENT_ID` (required)
- `DISABLE_AUTH` (default `false`)
- `NO_JWT_VALIDATION` (default `false`)
- `PERMIT_IO_API_KEY` (optional but required for full Permit.io operation)
- `PERMIT_IO_PDP_URL` (default `http://localhost:7766`)
Auth path customization:
- `COGNITO_LOGIN_PATH` (default `/login`)
- `COGNITO_CALLBACK_PATH` (default `/login-callback`)
- `COGNITO_HOME_PATH` (default `/home`)
- `COGNITO_LOGOUT_PATH` (default `/logout`)
- `HEALTH_PATH` (default `/health`)
## Rate Limiting
From `internal/serviceconfig/ratelimit/config.go`:
- `RATE_LIMIT_GLOBAL_RATE`
- `RATE_LIMIT_GLOBAL_BURST`
- `RATE_LIMIT_DEFAULT_RATE`
- `RATE_LIMIT_DEFAULT_BURST`
- `RATE_LIMIT_EXPIRES_IN`
- `RATE_LIMIT_ENDPOINT_OVERRIDES` (JSON)
Defaults:
- Global: `500 rps`, burst `1000`
- Per-IP: `5 rps`, burst `10`
## Logging and Observability
- `LOG_LEVEL` (default `INFO`)
- `ENABLE_OTEL` (default `false`)
## Notes
- Some deployment files still include legacy query/text environment variables because deprecated compatibility services remain in compose definitions.
- For API/runtime behavior, prefer `serviceAPIs/queryAPI.yaml` plus `internal/serviceconfig/*` as source of truth.
The system has expanded configuration support for batch processing, enhanced observability, and Permit.io RBAC integration.
## Configuration Architecture
The system implements a hierarchical, environment-driven configuration system designed for flexibility, type safety, and testability. Configuration is managed through the `internal/serviceconfig` package with domain-specific sub-packages.
## Configuration Hierarchy
### Environment Variable Priority (First Wins)
1. **devbox.json "env"** section - Highest priority for development
2. **devbox.json "init_hook"** exports - Build-time configuration
3. **.env file** variables - Local overrides
4. **System environment** - Runtime environment variables
### Configuration Loading Process
```go
// 1. Load .env file if present (development)
godotenv.Load()
// 2. Parse environment variables into struct
env.Parse(&config)
// 3. Validate required fields
config.Validate()
// 4. Initialize dependent services
config.Initialize(ctx)
```
## Base Configuration Structure
### Core Configuration (`serviceconfig/common.go`)
```go
type BaseConfig struct {
// Application
LogLevel string `env:"LOG_LEVEL" envDefault:"INFO"`
// Server
Port int `env:"PORT" envDefault:"8080"`
HTTPHost string `env:"HTTP_HOST" envDefault:"0.0.0.0"`
// Feature Flags
DisableAuth bool `env:"DISABLE_AUTH" envDefault:"false"`
EnableOTEL bool `env:"ENABLE_OTEL" envDefault:"false"`
// Observability
MetricsPath string `env:"METRICS_PATH" envDefault:"/metrics"`
}
```
### Configuration Providers Pattern
```go
// Interface composition for service dependencies
type ConfigProvider interface {
GetLogLevel() string
GetPort() int
GetHTTPHost() string
IsAuthDisabled() bool
}
// Service-specific configuration embeds base
type ServiceConfig struct {
BaseConfig
DatabaseConfig database.Config
AWSConfig aws.Config
QueueConfig queue.Config
}
```
## Domain-Specific Configuration
### Database Configuration (`serviceconfig/database/`)
```go
type Config struct {
// Connection
Host string `env:"PGHOST" envDefault:"localhost"`
Port int `env:"PGPORT" envDefault:"5432"`
Database string `env:"PGDATABASE"`
Username string `env:"PGUSER"`
Password string `env:"PGPASSWORD"`
// Connection Pool
MaxOpenConns int `env:"DB_MAX_OPEN_CONNS" envDefault:"25"`
MaxIdleConns int `env:"DB_MAX_IDLE_CONNS" envDefault:"25"`
ConnMaxLifetime int `env:"DB_CONN_MAX_LIFETIME" envDefault:"300"`
// Security
SSLMode string `env:"DB_SSL_MODE" envDefault:"require"`
NoSSL bool `env:"DB_NOSSL" envDefault:"false"`
}
// Provider interface
type ConfigProvider interface {
GetDatabaseConfig() Config
}
// Connection pool management
func (c Config) CreatePool(ctx context.Context) (*pgxpool.Pool, error) {
dsn := c.buildConnectionString()
return pgxpool.New(ctx, dsn)
}
```
### AWS Configuration (`serviceconfig/aws/`)
```go
type Config struct {
// Credentials
Region string `env:"AWS_REGION" envDefault:"us-east-1"`
AccessKeyID string `env:"AWS_ACCESS_KEY_ID"`
SecretAccessKey string `env:"AWS_SECRET_ACCESS_KEY"`
SessionToken string `env:"AWS_SESSION_TOKEN"`
Profile string `env:"AWS_PROFILE"`
// Endpoints (for LocalStack/testing)
EndpointURL string `env:"AWS_ENDPOINT_URL"`
S3UsePathStyle bool `env:"AWS_S3_USE_PATH_STYLE" envDefault:"false"`
// Service-specific
S3Bucket string `env:"S3_BUCKET"`
TextractRoleARN string `env:"TEXTRACT_ROLE_ARN"`
}
// SDK integration
func (c Config) LoadAWSConfig(ctx context.Context) (aws.Config, error) {
var opts []func(*config.LoadOptions) error
if c.Region != "" {
opts = append(opts, config.WithRegion(c.Region))
}
if c.EndpointURL != "" {
opts = append(opts, config.WithEndpointResolverWithOptions(
aws.EndpointResolverWithOptionsFunc(func(service, region string, options ...interface{}) (aws.Endpoint, error) {
return aws.Endpoint{URL: c.EndpointURL}, nil
}),
))
}
return config.LoadDefaultConfig(ctx, opts...)
}
```
### Queue Configuration (`serviceconfig/queue/`)
```go
// Base queue configuration with batch processing support
type Config struct {
// Connection
QueueURL string `env:"QUEUE_URL"`
MaxMessages int32 `env:"QUEUE_MAX_MESSAGES" envDefault:"10"`
WaitTimeSeconds int32 `env:"QUEUE_WAIT_TIME" envDefault:"20"`
VisibilityTimeout int32 `env:"QUEUE_VISIBILITY_TIMEOUT" envDefault:"30"`
// Processing (enhanced for batch operations)
WorkerCount int `env:"QUEUE_WORKER_COUNT" envDefault:"5"`
MaxRetries int `env:"QUEUE_MAX_RETRIES" envDefault:"3"`
RetryDelay int `env:"QUEUE_RETRY_DELAY" envDefault:"5"`
BatchSize int `env:"QUEUE_BATCH_SIZE" envDefault:"10"`
}
// Service-specific queue configs
type DocumentInitConfig struct {
Config
DocumentSyncURL string `env:"DOCUMENT_SYNC_URL"`
}
type QueryConfig struct {
Config
QuerySyncURL string `env:"QUERY_SYNC_URL"`
}
```
### Authentication Configuration (`serviceconfig/auth/`)
```go
type Config struct {
// Cognito Configuration
UserPoolID string `env:"COGNITO_USER_POOL_ID"`
ClientID string `env:"COGNITO_CLIENT_ID"`
ClientSecret string `env:"COGNITO_CLIENT_SECRET" envSecret:"true"`
Domain string `env:"COGNITO_DOMAIN"`
// URLs
RedirectURL string `env:"COGNITO_REDIRECT_URL"`
LogoutURL string `env:"COGNITO_LOGOUT_URL"`
// JWT Configuration
JWKSCacheExpiry time.Duration `env:"JWKS_CACHE_EXPIRY" envDefault:"1h"`
TokenExpiry time.Duration `env:"TOKEN_EXPIRY" envDefault:"24h"`
// Feature Flags
DisableAuth bool `env:"DISABLE_AUTH" envDefault:"false"`
}
// Dynamic URL generation
func (c Config) GetAuthorizationURL() string {
return fmt.Sprintf("https://%s.auth.%s.amazoncognito.com/oauth2/authorize",
c.Domain, extractRegion(c.UserPoolID))
}
func (c Config) GetTokenURL() string {
return fmt.Sprintf("https://%s.auth.%s.amazoncognito.com/oauth2/token",
c.Domain, extractRegion(c.UserPoolID))
}
```
## Environment Variables Reference
### Core Application Variables
```bash
# Logging and Debugging
LOG_LEVEL=DEBUG|INFO|WARN|ERROR # Default: INFO
ENABLE_OTEL=true|false # Default: false
# Server Configuration
PORT=8080 # Default: 8080
HTTP_HOST=0.0.0.0 # Default: 0.0.0.0
METRICS_PATH=/metrics # Default: /metrics
# Feature Flags
DISABLE_AUTH=true|false # Default: false (enable for local dev)
```
### Database Variables
```bash
# Connection
PGHOST=localhost # Database host
PGPORT=5432 # Database port
PGDATABASE=queryorchestration # Database name
PGUSER=postgres # Database user
PGPASSWORD=password # Database password
# Connection Pool
DB_MAX_OPEN_CONNS=25 # Default: 25
DB_MAX_IDLE_CONNS=25 # Default: 25
DB_CONN_MAX_LIFETIME=300 # Seconds, Default: 300
# Security
DB_SSL_MODE=require|disable # Default: require
DB_NOSSL=true|false # Default: false (for local dev)
```
### AWS Configuration Variables
```bash
# Credentials
AWS_REGION=us-east-1 # AWS region
AWS_ACCESS_KEY_ID=your-access-key # AWS access key
AWS_SECRET_ACCESS_KEY=your-secret # AWS secret key
AWS_SESSION_TOKEN=session-token # Optional session token
AWS_PROFILE=default # Optional AWS profile
# LocalStack/Testing
AWS_ENDPOINT_URL=http://localstack:4566 # Override AWS endpoints
AWS_S3_USE_PATH_STYLE=true|false # S3 path style (LocalStack)
# Service Configuration
S3_BUCKET=documents-bucket # S3 bucket for documents
TEXTRACT_ROLE_ARN=arn:aws:iam::... # Textract service role
```
### Queue Configuration Variables
```bash
# Queue URLs (SQS)
STORE_EVENT_URL=https://sqs.region.amazonaws.com/account/store-events
DOCUMENT_INIT_URL=https://sqs.region.amazonaws.com/account/doc-init
DOCUMENT_SYNC_URL=https://sqs.region.amazonaws.com/account/doc-sync
DOCUMENT_CLEAN_URL=https://sqs.region.amazonaws.com/account/doc-clean
DOCUMENT_TEXT_URL=https://sqs.region.amazonaws.com/account/doc-text
QUERY_SYNC_URL=https://sqs.region.amazonaws.com/account/query-sync
QUERY_URL=https://sqs.region.amazonaws.com/account/query
CLIENT_SYNC_URL=https://sqs.region.amazonaws.com/account/client-sync
QUERY_VERSION_SYNC_URL=https://sqs.region.amazonaws.com/account/query-version-sync
# Queue Processing
QUEUE_MAX_MESSAGES=10 # Messages per batch
QUEUE_WAIT_TIME=20 # Long polling seconds
QUEUE_VISIBILITY_TIMEOUT=30 # Message visibility
QUEUE_WORKER_COUNT=5 # Concurrent workers
QUEUE_MAX_RETRIES=3 # Retry attempts
QUEUE_RETRY_DELAY=5 # Retry delay seconds
```
### Authentication Variables
```bash
# AWS Cognito
COGNITO_USER_POOL_ID=us-east-1_xxxxxxx # Cognito User Pool ID
COGNITO_CLIENT_ID=client-id # Cognito App Client ID
COGNITO_CLIENT_SECRET=client-secret # Cognito App Client Secret
COGNITO_DOMAIN=your-domain # Cognito domain name
# URLs
COGNITO_REDIRECT_URL=http://localhost:8080/login-callback
COGNITO_LOGOUT_URL=http://localhost:8080/logout
# JWT Settings
JWKS_CACHE_EXPIRY=1h # JWKS cache duration
TOKEN_EXPIRY=24h # JWT token expiry
```
### Authorization Variables (Permit.io RBAC)
```bash
# Permit.io API Configuration
PERMIT_IO_API_KEY=permit_key_xxxxxxx # Permit.io API key (project/env derived from key)
PERMIT_IO_TENANT=default # Permit.io tenant identifier (default: "default")
PERMIT_IO_PDP_URL=http://localhost:7766 # Policy Decision Point URL (optional, uses cloud if not set)
PERMIT_IO_BASE_URL=https://api.permit.io # Permit.io API base URL (optional, for testing)
```
**Note:** The Permit.io project ID and environment ID are automatically derived from the API key
using the `/v2/api-key/scope` endpoint. Use a project-level or environment-level API key;
organization-level API keys are not supported.
## Configuration Validation
### Required Field Validation
```go
type Config struct {
DatabaseURL string `env:"DATABASE_URL,required"`
S3Bucket string `env:"S3_BUCKET,required"`
QueueURL string `env:"QUEUE_URL,required"`
}
// Validation on startup
func (c *Config) Validate() error {
var errors []string
if c.DatabaseURL == "" {
errors = append(errors, "DATABASE_URL is required")
}
if c.S3Bucket == "" {
errors = append(errors, "S3_BUCKET is required")
}
if len(errors) > 0 {
return fmt.Errorf("configuration validation failed: %s", strings.Join(errors, ", "))
}
return nil
}
```
### Secret Masking
```go
// Automatic masking of sensitive values in logs
type Config struct {
DatabasePassword string `env:"PGPASSWORD" envSecret:"true"`
AWSSecretKey string `env:"AWS_SECRET_ACCESS_KEY" envSecret:"true"`
JWTSecret string `env:"JWT_SECRET" envSecret:"true"`
}
// String() method automatically masks secrets
func (c Config) String() string {
return env.String(c) // Automatically masks fields tagged with envSecret
}
```
## Development vs Production Configuration
### Development Environment (.env)
```bash
# Local development overrides
LOG_LEVEL=DEBUG
DISABLE_AUTH=true
DB_NOSSL=true
AWS_ENDPOINT_URL=http://localstack:4566
AWS_S3_USE_PATH_STYLE=true
# Local service URLs
PGHOST=localhost
PGPORT=5432
```
### Production Environment
```bash
# Production settings
LOG_LEVEL=INFO
DISABLE_AUTH=false
DB_SSL_MODE=require
# Production AWS
AWS_REGION=us-east-1
# Credentials from IAM roles or environment
# Production database
PGHOST=production-db.region.rds.amazonaws.com
PGPORT=5432
DB_SSL_MODE=require
```
### Testing Environment
```bash
# Test-specific overrides
LOG_LEVEL=WARN
DISABLE_AUTH=true
DB_NOSSL=true
# Test containers
PGHOST=test-container
AWS_ENDPOINT_URL=http://test-localstack:4566
```
## Configuration Best Practices
### Security
- **Never commit secrets**: Use environment variables or secret management
- **Mask sensitive logs**: Use `envSecret` tag for automatic masking
- **Validate inputs**: Check configuration values at startup
- **Use least privilege**: Configure minimal required permissions
### Maintainability
- **Default values**: Provide sensible defaults for optional configuration
- **Documentation**: Document all environment variables and their purpose
- **Type safety**: Use strong typing and validation
- **Composition**: Build configuration from smaller, focused interfaces
### Testing
- **Override-friendly**: Allow easy configuration overrides for tests
- **Isolation**: Each test should have independent configuration
- **Realistic defaults**: Test configuration should mirror production patterns
- **Environment separation**: Clear separation between dev/test/prod configs
## Setting Up Authentication for a New Environment
This section covers the initial setup of authentication infrastructure when deploying to a new environment (e.g., dev, UAT, production). This includes configuring Permit.io for authorization and creating the initial bootstrap users in both AWS Cognito and Permit.io.
### Prerequisites
Before setting up authentication for a new environment, ensure you have:
1. **AWS Cognito User Pool** already created for the environment
2. **AWS credentials** configured with permissions to create Cognito users
3. **Access to Permit.io** organization account
### Step 1: Create Permit.io Environment and Get API Key
1. Log in to the [Permit.io Dashboard](https://app.permit.io)
2. Navigate to your project (or create a new one for the environment)
3. Create a new **Environment** for your deployment (e.g., `Development`, `UAT`, `Production`)
4. Go to **Settings → API Keys** and create a new API key scoped to this environment
5. Copy the API key - you will need it for the following steps
**Important**: Each environment (dev, UAT, prod) requires its own separate Permit.io environment and API key.
### Step 2: Clean Up Default Permit.io Roles
When you create a new environment in Permit.io, it automatically creates default `admin`, `viewer`, and `editor` roles. These must be deleted before running the setup tool because they conflict with the role names used by this system.
1. In the Permit.io Dashboard, navigate to **Policy → Roles**
2. Delete the default `admin`, `viewer`, and `editor` roles
3. Verify no roles remain before proceeding
### Step 3: Run the Permit.io Setup Tool
The permit setup tool configures all required resources and roles in Permit.io based on the `permit_policies.yaml` configuration file.
```bash
cd cmd/cognito_test/permit.setup
# Set the API key for your target environment
export PERMIT_API_KEY="permit_key_your_environment_key_here"
# Verify connection (list current configuration)
go run setup_permit.go -list
# Dry run to preview changes
go run setup_permit.go -config permit_policies.yaml -project default -env <your-env> -dry-run
# Apply the configuration
go run setup_permit.go -config permit_policies.yaml -project default -env <your-env>
```
This creates:
- All required **resources** (admin, client, document, folders, etc.) with their actions
- All required **roles** (super_admin, user_admin, auditor, editor, viewer, etc.) with appropriate permissions
See `cmd/cognito_test/permit.setup/README.md` for detailed documentation.
### Step 4: Create Bootstrap Users
After Permit.io is configured, use the user creation tool to create the initial users in both AWS Cognito and Permit.io. The tool ensures that users are created with matching Subject IDs in both systems, which is critical for authentication to work correctly.
**Important**: Always use the user creation tool to create users. Do not manually create users in Cognito and Permit.io separately, as this will result in mismatched Subject IDs and authentication failures.
```bash
cd cmd/cognito_test/user.creation.tool
# Configure environment variables
export PERMIT_KEY="permit_key_your_environment_key_here"
export COGNITO_USER_POOL_ID="us-east-1_xxxxxxxxx"
export AWS_REGION="us-east-1"
# Either use AWS credentials or an SSO profile
export AWS_PROFILE="your-sso-profile" # if using SSO
# Create a CSV file with your bootstrap users
# Format: email,first_name,last_name,role1,role2,...
cat > bootstrap_users.csv << 'EOF'
email,first_name,last_name,super_admin,user_admin
admin@yourcompany.com,Admin,User,super_admin,
useradmin@yourcompany.com,User,Admin,,user_admin
EOF
# Dry run to preview
go run main.go -project use-key -csv bootstrap_users.csv --dry-run
# Create the users
go run main.go -project use-key -csv bootstrap_users.csv
```
The tool will:
1. Create each user in AWS Cognito (with email as username)
2. Create corresponding user in Permit.io using the Cognito Subject ID as the key
3. Assign the specified roles in Permit.io
See `cmd/cognito_test/user.creation.tool/readme.md` for detailed documentation including:
- Full CSV format specification
- Delete, disable, and enable operations
- Audit logging and compliance features
- Troubleshooting guide
### Step 5: Verify Setup
After creating bootstrap users:
1. **Test Cognito login**: Verify users can authenticate through the Cognito hosted UI or your application
2. **Test API access**: Make an authenticated request to verify authorization works:
```bash
# Get a token (via Cognito login flow)
# Then test the identity endpoint
curl -H "Authorization: Bearer <access_token>" https://your-api/identity
```
3. **Check Permit.io Dashboard**: Verify users appear with correct role assignments
### Environment-Specific Scripts
For convenience, you can create environment-specific run scripts:
```bash
# Example: run.tool.prod.sh
#!/bin/bash
export PERMIT_KEY="permit_key_production_key_here"
export COGNITO_USER_POOL_ID="us-east-1_prodPoolId"
export AWS_REGION="us-east-1"
export AWS_PROFILE="production-profile"
go run main.go -project use-key -csv "$@"
```
### Quick Reference: New Environment Checklist
| Step | Action | Tool/Location |
|------|--------|---------------|
| 1 | Create Permit.io environment | Permit.io Dashboard |
| 2 | Get environment API key | Permit.io Dashboard → Settings → API Keys |
| 3 | Delete default roles | Permit.io Dashboard → Policy → Roles |
| 4 | Run permit setup | `cmd/cognito_test/permit.setup/` |
| 5 | Create bootstrap users | `cmd/cognito_test/user.creation.tool/` |
| 6 | Verify authentication | Test login and API access |
+12 -31
View File
@@ -12,13 +12,13 @@ This guide covers deployment, monitoring, and operational procedures for the enh
devbox shell
# 2. Start all services with dependencies
task dev:up
task compose:up
# 3. Verify deployment
task health:check
curl http://localhost:8080/health
# 4. View logs
task logs:follow
docker compose -f deployments/compose.local.yaml logs -f
```
### Production Deployment
@@ -55,11 +55,11 @@ kubectl rollout status deployment/query-api
# staging.env
LOG_LEVEL=DEBUG
ENABLE_OTEL=true
DB_SSL_MODE=require
DB_NOSSL=false
# Staging AWS resources
AWS_REGION=us-west-2
S3_BUCKET=doczy-staging-documents
BUCKET=doczy-staging-documents
```
**Production Environment**:
@@ -67,11 +67,11 @@ S3_BUCKET=doczy-staging-documents
# production.env
LOG_LEVEL=INFO
ENABLE_OTEL=true
DB_SSL_MODE=require
DB_NOSSL=false
# Production AWS resources
AWS_REGION=us-east-1
S3_BUCKET=doczy-production-documents
BUCKET=doczy-production-documents
```
### Rolling Updates
@@ -320,8 +320,8 @@ service:"queryAPI" AND latency > 1000ms | head 100
# Database connection issues
msg:"database connection" AND level:"ERROR"
# Queue processing bottlenecks
service:"queryRunner" | stats avg(processing_time) by query_type
# Queue processing bottlenecks (active pipeline)
service:"doc*Runner" | stats avg(processing_time) by service
```
### Debugging Common Issues
@@ -563,28 +563,9 @@ Queue consumer services require SQS permissions:
}
```
### Textract Permissions
### Legacy Textract Permissions
For OCR text extraction, additional Textract permissions are required:
```json
{
"Version": "2012-10-17",
"Statement": [
{
"Sid": "TextractPermissions",
"Effect": "Allow",
"Action": [
"textract:StartDocumentTextDetection",
"textract:GetDocumentTextDetection",
"textract:StartDocumentAnalysis",
"textract:GetDocumentAnalysis"
],
"Resource": "*"
}
]
}
```
The prior text extraction/query pipeline is deprecated. Keep Textract IAM permissions only if you still run legacy compatibility services in your deployment.
## Security Operations
@@ -626,4 +607,4 @@ grep -E "(POST|PUT|DELETE)" /var/log/nginx/access.log | awk '{print $1, $7, $9}'
# Database audit queries
SELECT * FROM pg_stat_user_tables WHERE n_tup_ins + n_tup_upd + n_tup_del > 0;
```
```
@@ -313,7 +313,6 @@ Permissions follow the format `resource:action`:
| `clients` | `get` | List all clients |
| `document` | `get`, `post`, `patch`, `delete` | Document operations |
| `folders` | `get`, `post`, `patch`, `delete` | Folder management |
| `query` | `get`, `post`, `patch` | Query definitions |
| `export` | `get`, `post` | Export operations |
| `admin` | `get`, `post`, `patch`, `delete` | User and EULA administration |
| `eula` | `get`, `post` | EULA status and agreement |
+2 -2
View File
@@ -84,13 +84,13 @@ The **prefix** portion determines the directory structure in S3:
The **filename** portion is delimited by the tilde character (`~`). It comes in two variants depending on how the document was uploaded:
**4-segment filename (single upload via `POST /clients/{id}/documents`):**
**4-segment filename (single upload via `POST /client/{id}/document`):**
```
{Timestamp}~{ClientID}~{Location}~{EntityID}[.FileType]
```
**5-segment filename (batch upload via `POST /clients/{id}/documents/batches`):**
**5-segment filename (batch upload via `POST /client/{id}/document/batch`):**
```
{Timestamp}~{ClientID}~{Location}~{EntityID}~{BatchID}[.FileType]
+8 -8
View File
@@ -15,7 +15,7 @@ This documentation provides comprehensive coverage of the DoczyAI Document Proce
- Deployment architecture and scalability
2. **[Service Architecture](02-service-architecture.md)**
- Detailed service catalog (8 services)
- Detailed service catalog (active + compatibility services)
- Document processing pipeline flow
- Inter-service communication patterns
- Queue architecture and message formats
@@ -63,7 +63,7 @@ This documentation provides comprehensive coverage of the DoczyAI Document Proce
- Token types and refresh strategy
- CORS configuration
10. **[Appendix](10-appendex.md)**
10. **[Appendix](10-appendix.md)**
- S3 Storage Guide: BucketKey path format, segment reference, partitioning
- Watching S3 for client document imports
- Document recovery from S3 buckets
@@ -92,20 +92,20 @@ This documentation provides comprehensive coverage of the DoczyAI Document Proce
### 🏗️ Architecture
- **Event-driven microservices** with SQS-based communication
- **8 distinct services**: 1 HTTP API + 6 runners + 1 utility
- **Active services**: 1 HTTP API + 5 active runners
- **Compatibility services**: additional deprecated runner binaries may still be present in some deployments
- **Multi-tenant design** with client isolation
- **Comprehensive versioning** for configurations
### 🔧 Technology Stack
- **Backend**: Go 1.24.0 with Echo framework
- **Database**: PostgreSQL 17.2 with SQLC
- **Cloud**: AWS (S3, SQS, Textract, Cognito)
- **Cloud**: AWS (S3, SQS, Cognito)
- **Containers**: Docker with multi-stage builds
- **Monitoring**: Prometheus + OpenTelemetry
### 📊 Key Capabilities
- **Document Processing**: PDF upload, cleaning, and text extraction
- **Text Extraction**: Automated OCR with AWS Textract
- **Document Processing**: PDF upload, initialization, sync-gating, and cleaning
- **Multi-tenant Support**: Isolated client environments
- **Document Organization**: Folders and labels for document management
- **Export Operations**: Structured data export in multiple formats
@@ -246,10 +246,10 @@ This workflow ensures all documentation formats are current with proper diagram
- **Health Checks**: http://localhost:8080/health
### External Dependencies
- **AWS Services**: S3, SQS, Textract, Cognito
- **AWS Services**: S3, SQS, Cognito
- **Database**: PostgreSQL 17.2+
- **Monitoring**: Prometheus-compatible metrics
---
*This documentation is automatically generated and maintained alongside the codebase. For questions or improvements, please contact the platform team.*
*This documentation is automatically generated and maintained alongside the codebase. For questions or improvements, please contact the platform team.*
+7 -1
View File
@@ -365,9 +365,15 @@ Rate limit headers included in all responses:
- **Authentication**: JWT Bearer tokens via AWS Cognito OAuth2
- **Authorization**: Role-based access control (RBAC) via Permit.io
- **All endpoints require authentication** except:
- **Public endpoints** (no authentication required):
- `GET /login` - Public login initiation
- `GET /login-callback` - OAuth2 callback handler
- `GET /home` - Home/menu page
- `GET /logout` - Logout redirect flow
- `GET /health` - Health check
- `GET /swagger/*` - Swagger UI
- `GET /metrics` - Prometheus metrics
- `GET /eula` - Current public EULA version
## API Versioning
+72 -318
View File
@@ -1,362 +1,116 @@
# Service Processing Details
This document provides detailed information about what each service in the document processing pipeline actually does.
This document describes current service behavior in the processing pipeline.
## Document Processing Pipeline Services
## Active Pipeline Services
### 1. storeEventRunner
**Purpose**: Entry point for document processing, triggered by S3 upload events
**Purpose**: Entry point for upload events.
**Code Locations**:
- `cmd/storeEventRunner/` - Service entry point and main.go
- `api/storeEventRunner/` - Runner implementation and message handling
- `internal/document/store/` - Core document storage logic
- `internal/serviceconfig/queue/documentinit/` - Queue configuration for next stage
**Code locations**:
- `cmd/storeEventRunner/`
- `api/storeEventRunner/`
- `internal/document/store/`
**Processing Logic**:
- Receives S3 event notifications (ObjectCreated:Put, ObjectCreated:CompleteMultipartUpload)
- Validates that the event type is supported
- Parses the S3 object key to extract structured information (client ID, location, timestamps)
- Filters for objects in the "import" location only
- Extracts document metadata (bucket, key, hash/ETag)
- Sends message to DOCINIT queue with document location information
**Behavior**:
- Consumes S3 event notifications
- Validates supported event/object patterns
- Extracts bucket/key/hash metadata
- Publishes to `DOCINIT`
**Key Validations**:
- Only processes supported S3 events (Put and CompleteMultipartUpload)
- Only processes objects with "import" location in their key path
- Validates key can be parsed into expected structure
**Output**: Sends to DOCINIT queue with {bucket, key, hash}
**Output**: `{bucket, key, hash}` to `DOCINIT`.
---
### 2. docInitRunner
**Purpose**: Document registration and deduplication
**Purpose**: Document registration and deduplication.
**Code Locations**:
- `cmd/docInitRunner/` - Service entry point and main.go
- `api/docInitRunner/` - Runner implementation and message handling
- `internal/document/init/` - Core document initialization logic
- `internal/serviceconfig/queue/documentsync/` - Queue configuration for next stage
**Code locations**:
- `cmd/docInitRunner/`
- `api/docInitRunner/`
- `internal/document/init/`
**Processing Logic**:
- Parses the S3 key to extract client ID, batch ID, and other metadata
- Checks if document already exists by looking up hash in database for the client
- If document exists (duplicate):
- Adds new document entry to existing document record
- Does NOT trigger downstream processing (deduplication)
- If new document:
- Creates new document record in database with client ID and hash
- Associates batch ID if present
- Adds document entry with S3 location details
- Sends message to DOCSYNC queue to continue processing
**Behavior**:
- Parses upload metadata from S3 key
- Deduplicates by client/hash
- Creates document and entry records for new documents
- Publishes new documents to `DOCSYNC`
**Key Features**:
- Hash-based deduplication per client
- Maintains multiple entries for same document (different uploads)
- Transaction-based database operations for consistency
**Output**: Sends to DOCSYNC queue with {document_id} (only for new documents)
**Output**: `{document_id}` to `DOCSYNC` for new documents.
---
### 3. docSyncRunner
**Purpose**: Client synchronization validation
**Purpose**: Sync eligibility gate.
**Code Locations**:
- `cmd/docSyncRunner/` - Service entry point and main.go
- `api/docSyncRunner/` - Runner implementation and message handling
- `internal/document/sync/` - Core document synchronization logic
- `internal/client/` - Client management utilities
- `internal/document/` - Shared document utilities
- `internal/serviceconfig/queue/documentclean/` - Queue configuration for next stage
**Code locations**:
- `cmd/docSyncRunner/`
- `api/docSyncRunner/`
- `internal/document/sync/`
**Processing Logic**:
- Retrieves document summary from database
- Fetches client configuration
- Checks if client has "CanSync" flag enabled
- If CanSync is true:
- Sends document to DOCCLEAN queue for processing
- If CanSync is false:
- Logs that document won't be synced
- Processing stops here for this document
**Behavior**:
- Loads document + client state
- Checks client sync status
- Publishes eligible documents to `DOCCLEAN`
**Key Validations**:
- Client must exist in database
- Client must have CanSync=true to proceed
**Output**: Sends to DOCCLEAN queue with {document_id} (only if client can sync)
**Output**: `{document_id}` to `DOCCLEAN` when eligible.
---
### 4. docCleanRunner
**Purpose**: Document validation and preparation
**Purpose**: Document validation/cleaning stage.
**Code Locations**:
- `cmd/docCleanRunner/` - Service entry point and main.go
- `api/docCleanRunner/` - Runner implementation and message handling
- `internal/document/clean/` - Core document cleaning and validation logic
- `internal/document/types/` - Document type definitions and MIME type handling
- `internal/serviceconfig/queue/documenttext/` - Queue configuration for next stage
**Code locations**:
- `cmd/docCleanRunner/`
- `api/docCleanRunner/`
- `internal/document/clean/`
**Processing Logic**:
- Checks if document has already been cleaned (idempotency)
- If not cleaned:
- Retrieves document entry from database
- Makes HEAD request to S3 to get content metadata
- Validates content type/MIME type is acceptable (PDF)
- Downloads document from S3
- Validates document integrity (not corrupt)
- Stores clean record with:
- Success: bucket, key, hash, mimetype
- Failure: reason for failure (invalid MIME, corrupt file)
- Creates versioned clean entry for tracking
- Sends to DOCTEXT queue regardless of success/failure
**Behavior**:
- Idempotent clean processing
- Validates MIME/content constraints and S3-backed inputs
- Stores clean version entries
**Key Validations**:
- MIME type validation (must be acceptable format)
- Document corruption check
- Hash verification against S3 ETag
**Output**: Sends to DOCTEXT queue with {document_id}
**Output**: none. The active pipeline currently ends after clean.
---
### 5. docTextRunner
**Purpose**: Text extraction from documents using AWS Textract
### 5. clientSyncRunner
**Purpose**: Re-queue all documents for a client.
**Code Locations**:
- `cmd/docTextRunner/` - Service entry point and main.go
- `api/docTextRunner/` - Runner implementation and message handling
- `internal/document/text/` - Core text extraction logic using AWS Textract
- `internal/document/types/` - Document type utilities for text processing
- `internal/serviceconfig/textract/` - AWS Textract client configuration
- `internal/serviceconfig/queue/querysync/` - Queue configuration for next stage
**Code locations**:
- `cmd/clientSyncRunner/`
- `api/clientSyncRunner/`
- `internal/client/sync/`
Tests for this use mocks only. No tests to the actual aws textract service.
**Behavior**:
- Consumes `CLIENTSYNC`
- Enumerates client documents
- Publishes each to `DOCSYNC`
**Processing Logic**:
- Checks if text has already been extracted (idempotency)
- If not extracted:
- Retrieves clean document record
- Skips if document failed cleaning
- Downloads document from S3
- Determines page count
- For each page (in parallel using thread pool):
- Sends page to AWS Textract for OCR
- Processes base text blocks
- Identifies required features (tables, forms)
- Performs additional extraction if needed
- Merges text with page markers
- Stores extracted text in S3
- Records text extraction in database with versioning
- Sends to QUERYSYNC queue for query processing
**Output**: `{document_id}` to `DOCSYNC` for each client document.
**Key Features**:
- Parallel page processing for performance
- Supports tables and forms extraction
- Stores full text with page boundaries preserved
- Version tracking for text extractions
## Deprecated Compatibility Services
**Output**: Sends to QUERYSYNC queue with {document_id}
These services are retained for deployment compatibility and are currently health-only/no-op:
- `cmd/docTextRunner/`
- `cmd/querySyncRunner/`
- `cmd/queryRunner/`
- `cmd/queryVersionSyncRunner/`
---
## Active Data Flow Summary
### 6. querySyncRunner
**Purpose**: Query dependency resolution and scheduling
**Code Locations**:
- `cmd/querySyncRunner/` - Service entry point and main.go
- `api/querySyncRunner/` - Runner implementation and message handling
- `internal/query/sync/` - Core query synchronization logic
- `internal/query/result/sync/` - Result synchronization utilities
- `internal/serviceconfig/queue/query/` - Queue configuration for next stage
**Processing Logic**:
- Verifies document has extracted text
- Queries database for unsynced queries that:
- Are configured for this client (via collector)
- Have no dependencies OR
- Have all dependencies already satisfied
- For each eligible query:
- Sends message to QUERY queue for execution
**Key Features**:
- Dependency graph resolution
- Only schedules queries with satisfied dependencies
- Client-specific query configuration via collectors
**Output**: Sends to QUERY queue with {document_id, query_id} for each eligible query
---
### 7. queryRunner
**Purpose**: Query execution and result storage
**Code Locations**:
- `cmd/queryRunner/` - Service entry point and main.go
- `api/queryRunner/` - Runner implementation and message handling
- `internal/query/result/set/` - Core query result processing
- `internal/query/result/` - Query execution logic and processors
- `internal/query/result/processor/` - Base processor interface and utilities
- `internal/query/types/contextFull/` - CONTEXT_FULL query type implementation
- `internal/query/types/jsonExtractor/` - JSON_EXTRACTOR query type implementation
- `internal/query/` - Query management utilities
- `internal/query/result/sync/` - Result synchronization for triggering dependents
**Processing Logic**:
- Retrieves document text version information
- Fetches query definition and active version
- If query has dependencies:
- Retrieves results from required queries
- Validates all dependencies are satisfied
- Executes query based on type:
- **CONTEXT_FULL**: Returns complete document text
- **JSON_EXTRACTOR**: Extracts specific JSON paths from dependent results
- Stores result in database with:
- Query result value
- Query version used
- Text version processed
- Dependencies tracked
- Identifies queries that depend on this result
- Sends dependent queries to QUERY queue (recursive processing)
**Key Features**:
- Supports multiple query types with different processors
- Dependency tracking and validation
- Recursive processing for dependent queries
- Version tracking for reproducibility
**Output**: Sends to QUERY queue with {document_id, query_id} for each dependent query
---
## Additional Synchronization Services
### 8. clientSyncRunner
**Purpose**: Client-level document synchronization
**Code Locations**:
- `cmd/clientSyncRunner/` - Service entry point and main.go
- `api/clientSyncRunner/` - Runner implementation and message handling
- `internal/client/sync/` - Core client synchronization logic
- `internal/serviceconfig/queue/documentsync/` - Queue configuration for document sync
**Processing Logic**:
- Processes CLIENTSYNC queue messages
- Retrieves all documents for a specific client
- Sends each document to DOCSYNC queue for processing
- Enables bulk reprocessing of client documents
**Output**: Sends to DOCSYNC queue with {document_id} for each client document
---
### 9. queryVersionSyncRunner
**Purpose**: Query version change propagation
**Code Locations**:
- `cmd/queryVersionSyncRunner/` - Service entry point and main.go
- `api/queryVersionSyncRunner/` - Runner implementation and message handling
- `internal/query/versionsync/` - Core query version synchronization logic
- `internal/serviceconfig/queue/clientsync/` - Queue configuration for client sync
**Processing Logic**:
- Processes QUERYVERSIONSYNC queue messages triggered by query updates
- Identifies all clients affected by query version changes
- Sends affected clients to CLIENTSYNC queue for reprocessing
- Enables cascading updates when query logic changes
**Output**: Sends to CLIENTSYNC queue with {client_id} for each affected client
---
## Data Flow Summary
### Primary Document Processing Pipeline
```
```text
S3 Upload Event
storeEventRunner (validates event, extracts metadata)
docInitRunner (deduplication, document registration)
docSyncRunner (client sync validation)
docCleanRunner (document validation, MIME check)
docTextRunner (OCR via AWS Textract)
querySyncRunner (find executable queries)
queryRunner (execute queries, trigger dependents)
queryRunner (recursive for dependent queries)
-> storeEventRunner
-> docInitRunner
-> docSyncRunner
-> docCleanRunner
```
### Synchronization Flows
```
Query Version Update
queryVersionSyncRunner (find affected clients)
clientSyncRunner (reprocess all client documents)
docSyncRunner → ... (normal processing pipeline)
```text
Client Sync Trigger
-> clientSyncRunner
-> docSyncRunner
-> docCleanRunner
```
## Key Design Patterns
### Idempotency
- Each service checks if its work has already been done
- Prevents duplicate processing on retries
- Version tracking ensures consistency
### Dependency Management
- Queries can depend on other query results
- Dependency graph is resolved before execution
- Results are chained through recursive processing
### Version Tracking
- Document clean versions
- Text extraction versions
- Query versions
- Enables reprocessing when configurations change
### Error Handling
- Failed documents are marked but continue through pipeline
- Allows partial processing and debugging
- Failure reasons are stored for analysis
### Client Isolation
- All processing is scoped to client ID
- Deduplication is per-client
- Query configurations are per-client via collectors
---
## Shared Code Directories
These directories contain utilities and configurations used across multiple services:
### Core Infrastructure
- `internal/database/` - Database queries, migrations, and repository patterns
- `internal/serviceconfig/` - Configuration management for all services
- `internal/serviceconfig/queue/` - SQS queue client configurations
- `internal/serviceconfig/objectstore/` - S3 client configurations
- `internal/serviceconfig/aws/` - AWS service configurations
- `internal/server/runner/` - Base runner framework for queue-based services
### Business Logic Libraries
- `internal/document/` - General document utilities (shared across document services)
- `internal/client/` - Client management utilities
- `internal/query/` - Query management and utilities (shared across query services)
### Testing & Utilities
- `internal/test/` - Test utilities and helpers
- `internal/validation/` - Input validation utilities
- `mocks/` - Auto-generated mock interfaces for testing
### API Layer
- `api/queryAPI/` - REST API handlers and controllers (not a queue-based runner)
- `pkg/queryAPI/` - Generated API client libraries
+1 -1
View File
@@ -619,7 +619,7 @@ func TestService_GetSummary(t *testing.T) {
```bash
task lint # Check for issues
task test:unit # Run unit tests
task test:functional # Run test suite
task fullsuite:ci # Run full test suite
```
+58
View File
@@ -0,0 +1,58 @@
# PDF Validation Requirements
## Overview
Before documents are sent to AWS Textract for text extraction, the system performs several validation checks to ensure the PDF files meet Textract's requirements and can be processed successfully.
## Validation Checks
### 1. File Type
**Requirement**: Only PDF files are accepted.
**What we check**: The file must have the MIME type `application/pdf`.
### 2. File Readability
**Requirement**: The PDF must be valid and readable.
**What we check**: The system attempts to open and parse the PDF file. Corrupted or malformed PDFs will be rejected.
### 3. File Size
**Requirement**: Each page must be 5 MB or smaller.
**What we check**: Every page in the PDF is validated individually. If any single page exceeds 5 MB, the document is rejected.
**Why this matters**: This is a requirement imposed by AWS Textract for synchronous processing.
### 4. Page Dimensions
**Minimum**: 50 x 50 pixels
**Maximum**: 10,000 x 10,000 pixels
**What we check**: Each page's width and height must fall within this range.
**Why this matters**: Pages that are too small may not contain readable content. Pages that are too large exceed Textract's processing limits.
### 5. Image Resolution (DPI)
**Minimum**: 72 DPI (dots per inch)
**What we check**: Each page must have a resolution of at least 72 DPI for both width and height.
**Why this matters**: Low-resolution documents may not produce accurate text extraction results. Documents scanned or created at lower resolutions should be re-scanned or regenerated at a higher quality.
## Validation Failure Reasons
When a document fails validation, the system records one of the following failure reasons:
- `invalid_mimetype` - File is not a PDF
- `invalid_read` - PDF is corrupted or cannot be opened
- `invalid_read_pages` - One or more pages cannot be read
- `zero_page_count` - PDF contains no pages
- `large_file` - One or more pages exceeds 5 MB
- `small_dimensions` - Page dimensions are too small (below 50x50 pixels)
- `large_dimensions` - Page dimensions are too large (above 10,000x10,000 pixels)
- `small_dpi` - Page resolution is below 72 DPI
- `large_dpi` - Page resolution is too high (future use)
+1 -1
View File
@@ -14,7 +14,7 @@ able to run on a scratch container.
To run the tests just run `./test.sh` from this directory.
Also the local permit.io pdp must also be running.
This can be started by running `demo.sh` in
`/cmd/cognito_test/permit_test/permit.harness`
`/cmd/auth_related/permit_test/permit.harness`
## Obtaining a reusable JWT token
If you observe the log output of running test.sh and look for
+3 -1
View File
@@ -266,9 +266,11 @@ func getCodeVerifier(state string, logger *slog.Logger) (string, error) {
func ExtractUserInfo(claims map[string]interface{}) UserInfo {
userInfo := UserInfo{}
// Extract username
// Extract username. ID tokens use "cognito:username", access tokens use "username".
if username, ok := claims["cognito:username"].(string); ok {
userInfo.Username = username
} else if username, ok := claims["username"].(string); ok {
userInfo.Username = username
}
// Extract email
+26
View File
@@ -399,6 +399,32 @@ func TestExtractUserInfo(t *testing.T) {
Groups: nil,
},
},
{
name: "access token claims with username instead of cognito:username",
claims: map[string]interface{}{
"sub": "e12bb510-30e1-7062-a69a-ca7f3f38d80e",
"username": "accesstokenuser",
"token_use": "access",
},
want: UserInfo{
Username: "accesstokenuser",
Email: "",
Groups: nil,
},
},
{
name: "cognito:username takes precedence over username",
claims: map[string]interface{}{
"cognito:username": "idtokenuser",
"username": "accesstokenuser",
"email": "user@example.com",
},
want: UserInfo{
Username: "idtokenuser",
Email: "user@example.com",
Groups: nil,
},
},
}
for _, tt := range tests {
+109
View File
@@ -0,0 +1,109 @@
#!/bin/bash
# Check if client ID parameter is provided
if [ -z "$1" ]; then
echo "Usage: $0 <client_id>"
echo "Example: $0 test_client_1758574881"
exit 1
fi
CLIENT_ID="$1"
# API configuration
# local host version of the test
#BASE_URL="http://localhost:8080"
# cloud version http://queryo-query-rtxkppcxpjsb-742555588.us-east-2.elb.amazonaws.com/
BASE_URL="http://queryo-query-rtxkppcxpjsb-742555588.us-east-2.elb.amazonaws.com"
ENDPOINT="/client/${CLIENT_ID}/document"
# Color codes for output
GREEN='\033[0;32m'
RED='\033[0;31m'
YELLOW='\033[0;33m'
BLUE='\033[0;34m'
CYAN='\033[0;36m'
NC='\033[0m' # No Color
echo "================================================"
echo -e "${BLUE}Document List for Client${NC}"
echo "================================================"
echo -e "Client ID: ${YELLOW}$CLIENT_ID${NC}"
echo -e "API Endpoint: ${CYAN}${BASE_URL}${ENDPOINT}${NC}"
echo "================================================"
echo ""
# Make API request
echo -e "${BLUE}Fetching documents...${NC}"
RESPONSE=$(curl -s -w "\n%{http_code}" "${BASE_URL}${ENDPOINT}")
# Extract HTTP status code (last line)
HTTP_STATUS=$(echo "$RESPONSE" | tail -n1)
# Extract response body (all lines except last) - using sed for portability
RESPONSE_BODY=$(echo "$RESPONSE" | sed '$d')
# Check HTTP status
if [ "$HTTP_STATUS" -eq 200 ]; then
echo -e "${GREEN}✓ API Request Successful${NC}"
echo ""
# Check if jq is available for JSON parsing
if command -v jq >/dev/null 2>&1; then
# Parse and display with jq
DOCUMENT_COUNT=$(echo "$RESPONSE_BODY" | jq '. | length' 2>/dev/null || echo "0")
if [ -z "$DOCUMENT_COUNT" ] || [ "$DOCUMENT_COUNT" = "null" ]; then
DOCUMENT_COUNT=0
fi
if [ "$DOCUMENT_COUNT" -eq 0 ]; then
echo -e "${YELLOW}No documents found for client: $CLIENT_ID${NC}"
else
echo -e "${GREEN}Found $DOCUMENT_COUNT document(s)${NC}"
echo ""
echo -e "${BLUE}Document Details:${NC}"
echo "------------------------------------------------"
# Display each document with formatting
echo "$RESPONSE_BODY" | jq -r '.[] |
"Document ID: \(.id)\n" +
"Hash: \(.hash)\n" +
"---"'
fi
else
# Fallback to basic formatting without jq
echo -e "${YELLOW}Note: jq not found, displaying raw JSON response${NC}"
echo ""
echo -e "${BLUE}Raw Response:${NC}"
echo "$RESPONSE_BODY" | python3 -m json.tool 2>/dev/null || echo "$RESPONSE_BODY"
fi
elif [ "$HTTP_STATUS" -eq 401 ]; then
echo -e "${RED}✗ Unauthorized (401)${NC}"
echo -e "${YELLOW}Authentication required. This endpoint requires JWT authentication.${NC}"
echo ""
echo -e "${BLUE}Response:${NC}"
echo "$RESPONSE_BODY"
elif [ "$HTTP_STATUS" -eq 404 ]; then
echo -e "${RED}✗ Not Found (404)${NC}"
echo -e "${YELLOW}Client '$CLIENT_ID' not found or no documents exist.${NC}"
echo ""
echo -e "${BLUE}Response:${NC}"
echo "$RESPONSE_BODY"
else
echo -e "${RED}✗ API Request Failed (HTTP $HTTP_STATUS)${NC}"
echo ""
echo -e "${BLUE}Response:${NC}"
echo "$RESPONSE_BODY"
fi
echo ""
echo "================================================"
# Exit with appropriate code
if [ "$HTTP_STATUS" -eq 200 ]; then
exit 0
else
exit 1
fi
+2 -2
View File
@@ -115,13 +115,13 @@ TEST_USER_DOMAIN=example.com
**Note:** The test suite automatically loads environment variables from `test/integration/.env` when you run the tests. You do NOT need to manually source the file.
Ensure the following roles exist in your Permit.io environment (as defined in `cmd/cognito_test/permit.setup/permit_policies.yaml`):
Ensure the following roles exist in your Permit.io environment (as defined in `cmd/auth_related/permit.setup/permit_policies.yaml`):
- `super_admin`
- `user_admin`
- `auditor`
- `client_user`
You can create roles through the Permit.io dashboard, API, or use the setup tool at `cmd/cognito_test/permit.setup/`.
You can create roles through the Permit.io dashboard, API, or use the setup tool at `cmd/auth_related/permit.setup/`.
## Running the Tests
+1 -1
View File
@@ -506,7 +506,7 @@ func (s *AdminIntegrationTestSuite) TestRoleAssignment() {
"To run this test, ensure the Permit.io environment has roles defined.\n" +
"Expected roles (from permit_policies.yaml):\n" +
" - super_admin\n - user_admin\n - auditor\n - client_user\n\n" +
"Run: task permit:setup (or use cmd/cognito_test/permit.setup/setup_permit)\n" +
"Run: task permit:setup (or use cmd/auth_related/permit.setup/setup_permit)\n" +
"Visit: https://app.permit.io")
return
}