HealthTech HIPAA-Compliant Pipeline

-- Table de faits principale pour les événements cliniques
CREATE TABLE FACT_CLINICAL_EVENT (
    event_sk BIGINT PRIMARY KEY AUTO_INCREMENT,
    event_id VARCHAR(50) NOT NULL,
    patient_research_id VARCHAR(100) NOT NULL, -- Hash unidirectionnel
    provider_sk INT NOT NULL,
    event_type_sk INT NOT NULL,
    event_timestamp TIMESTAMP(3) NOT NULL,
    date_sk INT NOT NULL,
    time_sk INT NOT NULL,
    severity_score DECIMAL(3,2),
    risk_score DECIMAL(3,2),
    clinical_values JSON, -- Valeurs cliniques chiffrées
    is_critical BOOLEAN DEFAULT FALSE,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    
    INDEX idx_patient_date (patient_research_id, date_sk),
    INDEX idx_provider_date (provider_sk, date_sk),
    INDEX idx_critical (is_critical) WHERE is_critical = TRUE
);

-- Table pour les résultats de laboratoire
CREATE TABLE FACT_LAB_RESULT (
    result_sk BIGINT PRIMARY KEY AUTO_INCREMENT,
    patient_research_id VARCHAR(100) NOT NULL,
    lab_test_sk INT NOT NULL,
    batch_id VARCHAR(50) NOT NULL,
    result_timestamp TIMESTAMP NOT NULL,
    numeric_value DECIMAL(10,4),
    text_value TEXT,
    unit VARCHAR(20),
    reference_range_low DECIMAL(10,4),
    reference_range_high DECIMAL(10,4),
    is_abnormal BOOLEAN DEFAULT FALSE,
    abnormal_flag VARCHAR(10),
    
    INDEX idx_patient_test (patient_research_id, lab_test_sk),
    INDEX idx_batch (batch_id),
    INDEX idx_abnormal (is_abnormal) WHERE is_abnormal = TRUE
);

-- Table pour l'administration de médicaments
CREATE TABLE FACT_MEDICATION_ADMINISTRATION (
    administration_sk BIGINT PRIMARY KEY AUTO_INCREMENT,
    patient_research_id VARCHAR(100) NOT NULL,
    medication_sk INT NOT NULL,
    administration_timestamp TIMESTAMP NOT NULL,
    dosage DECIMAL(10,4) NOT NULL,
    dosage_unit VARCHAR(20) NOT NULL,
    route VARCHAR(50), -- oral, IV, IM, etc.
    site VARCHAR(100), -- site d'injection si applicable
    administered_by_sk INT,
    is_scheduled BOOLEAN DEFAULT FALSE,
    
    INDEX idx_patient_medication (patient_research_id, medication_sk),
    INDEX idx_timestamp (administration_timestamp)
);

-- Dimension patient anonymisée
CREATE TABLE DIM_PATIENT (
    patient_sk INT PRIMARY KEY AUTO_INCREMENT,
    patient_research_id VARCHAR(100) UNIQUE NOT NULL,
    age_group VARCHAR(20), -- 18-25, 26-35, etc.
    gender_category VARCHAR(10), -- M, F, Other
    zip_code_prefix VARCHAR(3), -- 3 premiers chiffres seulement
    chronic_conditions_flags JSON, -- Flags booléens pour conditions
    risk_factors JSON, -- Facteurs de risque encodés
    registration_date DATE,
    last_activity_date DATE,
    valid_from TIMESTAMP NOT NULL,
    valid_to TIMESTAMP NULL,
    is_current BOOLEAN DEFAULT TRUE,
    
    INDEX idx_research_id (patient_research_id),
    INDEX idx_current (is_current)
);

-- Dimension diagnostic avec hiérarchie ICD-10
CREATE TABLE DIM_DIAGNOSIS (
    diagnosis_sk INT PRIMARY KEY AUTO_INCREMENT,
    icd10_code VARCHAR(10) NOT NULL,
    diagnosis_name VARCHAR(255) NOT NULL,
    chapter VARCHAR(100),
    section VARCHAR(100),
    category VARCHAR(100),
    subcategory VARCHAR(100),
    severity_level VARCHAR(20),
    is_chronic BOOLEAN DEFAULT FALSE,
    
    INDEX idx_icd10_code (icd10_code),
    INDEX idx_chapter (chapter),
    INDEX idx_chronic (is_chronic)
);

-- Dimension procédure avec codes CPT
CREATE TABLE DIM_PROCEDURE (
    procedure_sk INT PRIMARY KEY AUTO_INCREMENT,
    cpt_code VARCHAR(10) NOT NULL,
    procedure_name VARCHAR(255) NOT NULL,
    procedure_category VARCHAR(100),
    body_system VARCHAR(100),
    complexity_level VARCHAR(20),
    typical_duration_minutes INT,
    
    INDEX idx_cpt_code (cpt_code),
    INDEX idx_category (procedure_category)
);

-- Vue matérialisée pour la timeline patient
CREATE TABLE FACT_PATIENT_TIMELINE (
    timeline_sk BIGINT PRIMARY KEY AUTO_INCREMENT,
    patient_research_id VARCHAR(100) NOT NULL,
    timeline_date DATE NOT NULL,
    days_since_registration INT,
    active_medications_count INT,
    days_since_last_admission INT,
    vital_signs_trend JSON, -- Tendances des signes vitaux
    risk_score_daily DECIMAL(3,2),
    engagement_score DECIMAL(3,2),
    
    INDEX idx_patient_date (patient_research_id, timeline_date),
    INDEX idx_risk_score (risk_score_daily)
);

-- Feature store pour le ML médical
CREATE TABLE FEATURE_STORE_PATIENT (
    feature_sk BIGINT PRIMARY KEY AUTO_INCREMENT,
    patient_research_id VARCHAR(100) NOT NULL,
    feature_name VARCHAR(100) NOT NULL,
    feature_value JSON,
    feature_version VARCHAR(20),
    feature_timestamp TIMESTAMP NOT NULL,
    feature_source VARCHAR(100),
    feature_quality_score DECIMAL(3,2),
    
    INDEX idx_patient_feature (patient_research_id, feature_name),
    INDEX idx_timestamp (feature_timestamp)
);

-- Table pour les prédictions et recommandations
CREATE TABLE FACT_PREDICTION_RESULT (
    prediction_sk BIGINT PRIMARY KEY AUTO_INCREMENT,
    patient_research_id VARCHAR(100) NOT NULL,
    model_name VARCHAR(100) NOT NULL,
    model_version VARCHAR(20) NOT NULL,
    prediction_type VARCHAR(50), -- readmission, deterioration, etc.
    prediction_value DECIMAL(3,2),
    confidence_score DECIMAL(3,2),
    prediction_timestamp TIMESTAMP NOT NULL,
    actual_outcome VARCHAR(50), -- NULL si pas encore connu
    outcome_timestamp TIMESTAMP NULL,
    
    INDEX idx_patient_model (patient_research_id, model_name),
    INDEX idx_prediction_type (prediction_type)
);

// Configuration AWS HealthLake pour la conformité HIPAA
@Configuration
public class HealthLakeConfig {
    
    @Bean
    public HealthLakeClient healthLakeClient() {
        HealthLakeClientBuilder builder = HealthLakeClient.builder();
        builder.region(Region.US_EAST_1);
        builder.credentialsProvider(DefaultCredentialsProvider.create());
        
        return builder.build();
    }
    
    @Bean
    public FHIRService fhirService(HealthLakeClient healthLakeClient) {
        return new FHIRService(healthLakeClient);
    }
}

// Service FHIR pour l'ingestion des données médicales
@Service
public class FHIRService {
    
    private final HealthLakeClient healthLakeClient;
    private final DataAnonymizationService anonymizationService;
    
    public ImportJobResult importFHIRData(String dataLocation, DataFormat format) {
        try {
            // Configuration de l'import avec conformité HIPAA
            ImportJobRequest request = ImportJobRequest.builder()
                .inputDataConfig(InputDataConfig.builder()
                    .s3Uri(dataLocation)
                    .dataFormat(format)
                    .build())
                .jobOutputDataConfig(OutputDataConfig.builder()
                    .s3Uri("s3://healthlake-output/")
                    .build())
                .jobName("HIPAA-Import-" + System.currentTimeMillis())
                .dataAccessRoleArn("arn:aws:iam::account:role/HealthLakeDataAccessRole")
                .build();
            
            // Lancement de l'import
            ImportJobResponse response = healthLakeClient.startFHIRImportJob(request);
            
            // Monitoring de l'import
            return monitorImportJob(response.importJobId());
            
        } catch (HealthLakeException e) {
            throw new HIPAAComplianceException("Échec de l'import HealthLake", e);
        }
    }
}

// Configuration AWS IoT Core pour les wearables
@Configuration
public class IoTCoreConfig {
    
    @Bean
    public IoTCoreClient ioTCoreClient() {
        IoTCoreClientBuilder builder = IoTCoreClient.builder();
        builder.region(Region.US_EAST_1);
        builder.credentialsProvider(DefaultCredentialsProvider.create());
        
        return builder.build();
    }
    
    @Bean
    public KinesisClient kinesisClient() {
        KinesisClientBuilder builder = KinesisClient.builder();
        builder.region(Region.US_EAST_1);
        builder.credentialsProvider(DefaultCredentialsProvider.create());
        
        return builder.build();
    }
}

// Service de streaming des données IoT
@Service
public class IoTStreamingService {
    
    private final IoTCoreClient ioTCoreClient;
    private final KinesisClient kinesisClient;
    private final DataValidationService validationService;
    
    public void processWearableData(String deviceId, WearableData data) {
        try {
            // Validation des données IoT
            if (!validationService.validateWearableData(data)) {
                throw new InvalidDataException("Données wearables invalides");
            }
            
            // Enrichissement avec métadonnées HIPAA
            EnrichedData enrichedData = enrichWithHIPAAMetadata(data, deviceId);
            
            // Envoi vers Kinesis pour processing temps réel
            PutRecordRequest request = PutRecordRequest.builder()
                .streamName("wearable-data-stream")
                .partitionKey(deviceId)
                .data(SdkBytes.fromUtf8String(enrichedData.toJson()))
                .build();
            
            PutRecordResponse response = kinesisClient.putRecord(request);
            
            // Log de l'audit trail
            auditDataProcessing(deviceId, data, response.sequenceNumber());
            
        } catch (Exception e) {
            handleStreamingError(deviceId, data, e);
        }
    }
    
    private EnrichedData enrichWithHIPAAMetadata(WearableData data, String deviceId) {
        return EnrichedData.builder()
            .originalData(data)
            .deviceId(deviceId)
            .timestamp(Instant.now())
            .dataSource("wearable-device")
            .complianceLevel("HIPAA")
            .encryptionStatus("encrypted")
            .build();
    }
}

// Configuration SageMaker pour le ML médical
@Configuration
public class SageMakerConfig {
    
    @Bean
    public SageMakerClient sageMakerClient() {
        SageMakerClientBuilder builder = SageMakerClient.builder();
        builder.region(Region.US_EAST_1);
        builder.credentialsProvider(DefaultCredentialsProvider.create());
        
        return builder.build();
    }
    
    @Bean
    public FeatureStoreService featureStoreService(SageMakerClient sageMakerClient) {
        return new FeatureStoreService(sageMakerClient);
    }
}

// Service de feature store pour le ML médical
@Service
public class FeatureStoreService {
    
    private final SageMakerClient sageMakerClient;
    private final DataLineageService lineageService;
    
    public void createFeatureGroup(String featureGroupName, FeatureDefinition... definitions) {
        try {
            // Création du feature group avec conformité HIPAA
            CreateFeatureGroupRequest request = CreateFeatureGroupRequest.builder()
                .featureGroupName(featureGroupName)
                .recordIdentifierFeatureName("patient_research_id")
                .eventTimeFeatureName("timestamp")
                .featureDefinitions(Arrays.asList(definitions))
                .onlineStoreConfig(OnlineStoreConfig.builder()
                    .enableOnlineStore(true)
                    .build())
                .offlineStoreConfig(OfflineStoreConfig.builder()
                    .s3StorageConfig(S3StorageConfig.builder()
                        .s3Uri("s3://feature-store-offline/")
                        .build())
                    .build())
                .roleArn("arn:aws:iam::account:role/SageMakerFeatureStoreRole")
                .description("Feature group HIPAA-compliant pour " + featureGroupName)
                .build();
            
            CreateFeatureGroupResponse response = sageMakerClient.createFeatureGroup(request);
            
            // Configuration du monitoring des features
            configureFeatureMonitoring(featureGroupName);
            
            // Enregistrement de la lignée des données
            lineageService.recordFeatureGroupCreation(featureGroupName, definitions);
            
        } catch (SageMakerException e) {
            throw new MLPipelineException("Échec de création du feature group", e);
        }
    }
    
    private void configureFeatureMonitoring(String featureGroupName) {
        // Configuration du monitoring des features pour la conformité FDA
        MonitoringScheduleConfig monitoringConfig = MonitoringScheduleConfig.builder()
            .monitoringType(MonitoringType.DATA_QUALITY)
            .dataQualityJobInput(DataQualityJobInput.builder()
                .endpointInput(EndpointInput.builder()
                    .endpointName(featureGroupName + "-endpoint")
                    .build())
                .build())
            .build();
        
        // Création du schedule de monitoring
        CreateMonitoringScheduleRequest request = CreateMonitoringScheduleRequest.builder()
            .monitoringScheduleName(featureGroupName + "-monitoring")
            .monitoringScheduleConfig(monitoringConfig)
            .build();
        
        sageMakerClient.createMonitoringSchedule(request);
    }
}