Retail Legacy Migration Pipeline

-- Table de faits principale pour les ventes au niveau ligne
CREATE TABLE FACT_SALES_LINE (
    sale_line_sk BIGINT PRIMARY KEY AUTO_INCREMENT,
    sale_id VARCHAR(50) NOT NULL,
    store_sk INT NOT NULL,
    product_sk INT NOT NULL,
    customer_sk INT,
    date_sk INT NOT NULL,
    time_sk INT NOT NULL,
    quantity INT NOT NULL,
    unit_price DECIMAL(19,4) NOT NULL,
    discount_amount DECIMAL(19,4) DEFAULT 0,
    tax_amount DECIMAL(19,4) DEFAULT 0,
    net_amount DECIMAL(19,4) NOT NULL,
    channel_sk INT NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    
    INDEX idx_store_date (store_sk, date_sk),
    INDEX idx_product_date (product_sk, date_sk),
    INDEX idx_customer_date (customer_sk, date_sk)
);

-- Table pour les promotions multiples et empilables
CREATE TABLE FACT_SALES_PROMOTION_BRIDGE (
    sale_line_sk BIGINT NOT NULL,
    promotion_sk INT NOT NULL,
    promotion_amount DECIMAL(19,4) NOT NULL,
    promotion_type VARCHAR(50),
    promotion_priority INT,
    
    PRIMARY KEY (sale_line_sk, promotion_sk),
    FOREIGN KEY (sale_line_sk) REFERENCES FACT_SALES_LINE(sale_line_sk)
);

-- Snapshot quotidien des stocks par magasin
CREATE TABLE FACT_INVENTORY_SNAPSHOT (
    snapshot_sk BIGINT PRIMARY KEY AUTO_INCREMENT,
    store_sk INT NOT NULL,
    product_sk INT NOT NULL,
    date_sk INT NOT NULL,
    on_hand_quantity INT NOT NULL,
    on_order_quantity INT DEFAULT 0,
    allocated_quantity INT DEFAULT 0,
    available_to_promise INT GENERATED ALWAYS AS (on_hand_quantity - allocated_quantity) STORED,
    
    INDEX idx_store_product_date (store_sk, product_sk, date_sk),
    INDEX idx_date (date_sk)
);

-- Dimension produit avec hiérarchie profonde
CREATE TABLE DIM_PRODUCT (
    product_sk INT PRIMARY KEY AUTO_INCREMENT,
    product_id VARCHAR(50) NOT NULL,
    sku VARCHAR(100) UNIQUE NOT NULL,
    product_name VARCHAR(255) NOT NULL,
    brand_sk INT,
    category_sk INT,
    department_sk INT,
    division_sk INT,
    is_active BOOLEAN DEFAULT TRUE,
    launch_date DATE,
    discontinued_date DATE,
    
    INDEX idx_sku (sku),
    INDEX idx_category (category_sk),
    INDEX idx_active (is_active)
);

-- Modèle EAV pour les attributs spécifiques par catégorie
CREATE TABLE DIM_PRODUCT_ATTRIBUTE (
    product_sk INT NOT NULL,
    attribute_name VARCHAR(100) NOT NULL,
    attribute_value TEXT,
    attribute_type VARCHAR(50),
    
    PRIMARY KEY (product_sk, attribute_name),
    FOREIGN KEY (product_sk) REFERENCES DIM_PRODUCT(product_sk)
);

-- Dimension magasin avec hiérarchie organisationnelle
CREATE TABLE DIM_STORE (
    store_sk INT PRIMARY KEY AUTO_INCREMENT,
    store_id VARCHAR(50) NOT NULL,
    store_name VARCHAR(255) NOT NULL,
    store_number VARCHAR(20),
    surface_area DECIMAL(10,2),
    format_type VARCHAR(50), -- hypermarché, supermarché, proximité
    zone_chalandise VARCHAR(100),
    cluster_demographique VARCHAR(50),
    market_potential DECIMAL(15,2),
    market_share_index DECIMAL(5,4),
    performance_index DECIMAL(5,4),
    
    INDEX idx_store_id (store_id),
    INDEX idx_format (format_type),
    INDEX idx_cluster (cluster_demographique)
);

-- Dimension pour les événements impactant les ventes
CREATE TABLE DIM_CALENDAR_EVENT (
    event_sk INT PRIMARY KEY AUTO_INCREMENT,
    event_name VARCHAR(255) NOT NULL,
    event_type VARCHAR(100), -- holiday, vacances, événement local, météo
    event_date DATE NOT NULL,
    expected_impact VARCHAR(20), -- low, medium, high
    affected_radius_km INT,
    description TEXT,
    
    INDEX idx_event_date (event_date),
    INDEX idx_event_type (event_type)
);

-- Table de liaison pour les événements par magasin
CREATE TABLE BRIDGE_STORE_EVENT (
    store_sk INT NOT NULL,
    event_sk INT NOT NULL,
    impact_multiplier DECIMAL(5,4) DEFAULT 1.0,
    
    PRIMARY KEY (store_sk, event_sk),
    FOREIGN KEY (store_sk) REFERENCES DIM_STORE(store_sk),
    FOREIGN KEY (event_sk) REFERENCES DIM_CALENDAR_EVENT(event_sk)
);

-- Calendrier fiscal personnalisé (année commence en février)
CREATE TABLE DIM_FISCAL_CALENDAR (
    fiscal_date_sk INT PRIMARY KEY,
    fiscal_date DATE NOT NULL,
    fiscal_year INT NOT NULL,
    fiscal_quarter INT NOT NULL,
    fiscal_month INT NOT NULL,
    fiscal_week INT NOT NULL,
    is_fiscal_year_end BOOLEAN DEFAULT FALSE,
    is_fiscal_quarter_end BOOLEAN DEFAULT FALSE,
    
    INDEX idx_fiscal_date (fiscal_date),
    INDEX idx_fiscal_year (fiscal_year),
    INDEX idx_fiscal_quarter (fiscal_year, fiscal_quarter)
);

// Configuration SAP Data Services pour l'extraction
@Configuration
public class SAPDataServicesConfig {
    
    @Bean
    public DataSource sapDataSource() {
        HikariConfig config = new HikariConfig();
        config.setJdbcUrl("jdbc:sap://sap-server:3300/ECC6");
        config.setUsername("sap_username");
        config.setPassword("sap_password");
        config.setDriverClassName("com.sap.db.jdbc.Driver");
        config.setMaximumPoolSize(10);
        config.setMinimumIdle(5);
        
        return new HikariDataSource(config);
    }
    
    @Bean
    public SAPDataServices sapDataServices(DataSource dataSource) {
        SAPDataServicesConfig config = new SAPDataServicesConfig();
        config.setDataSource(dataSource);
        config.setExtractMode(ExtractMode.CDC); // Change Data Capture
        config.setBatchSize(10000);
        config.setExtractWindow("00:00-04:00"); // Fenêtre de maintenance
        
        return new SAPDataServices(config);
    }
}

// Service d'extraction avec gestion des erreurs
@Service
public class SAPExtractionService {
    
    private final SAPDataServices sapDataServices;
    private final ErrorHandlingService errorHandler;
    
    public ExtractionResult extractData(String tableName, LocalDateTime since) {
        try {
            // Extraction avec CDC sur les tables techniques SAP
            CDCConfig cdcConfig = new CDCConfig();
            cdcConfig.setChangeTables(Arrays.asList("CDHDR", "CDPOS"));
            cdcConfig.setSinceTimestamp(since);
            
            ExtractionResult result = sapDataServices.extractWithCDC(tableName, cdcConfig);
            
            // Validation des données extraites
            validateExtractedData(result);
            
            return result;
            
        } catch (SAPExtractionException e) {
            errorHandler.handleExtractionError(tableName, e);
            throw new MigrationException("Échec de l'extraction SAP", e);
        }
    }
}

// Configuration NiFi pour l'intégration multi-systèmes
@Configuration
public class NiFiConfig {
    
    @Bean
    public NiFiClient niFiClient() {
        NiFiClientConfig config = new NiFiClientConfig();
        config.setNiFiUrl("http://nifi:8080");
        config.setUsername("nifi_username");
        config.setPassword("nifi_password");
        config.setConnectionTimeout(30000);
        config.setReadTimeout(60000);
        
        return new NiFiClient(config);
    }
    
    @Bean
    public DataFlowManager dataFlowManager(NiFiClient niFiClient) {
        return new DataFlowManager(niFiClient);
    }
}

// Gestionnaire de flux de données pour différents formats
@Service
public class MultiSystemIntegrationService {
    
    private final DataFlowManager dataFlowManager;
    private final FormatConverterService formatConverter;
    
    public void createIntegrationFlow(String systemType, String format) {
        // Création d'un flux NiFi personnalisé selon le type de système
        DataFlow flow = new DataFlow();
        
        switch (systemType.toLowerCase()) {
            case "as400":
                flow.addProcessor(new AS400DataProcessor());
                flow.addProcessor(new EBCDICConverter());
                break;
                
            case "pos":
                flow.addProcessor(new POSDataProcessor());
                flow.addProcessor(new JSONConverter());
                break;
                
            case "legacy":
                flow.addProcessor(new LegacyDataProcessor());
                flow.addProcessor(new CSVConverter());
                break;
        }
        
        // Ajout des processeurs communs
        flow.addProcessor(new DataValidator());
        flow.addProcessor(new DataEnricher());
        flow.addProcessor(new KafkaProducer());
        
        // Déploiement du flux
        dataFlowManager.deployFlow(flow);
    }
}

// Configuration Delta Lake pour le Data Lakehouse
@Configuration
public class DeltaLakeConfig {
    
    @Bean
    public SparkSession sparkSession() {
        return SparkSession.builder()
            .appName("Retail Data Lakehouse")
            .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
            .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
            .config("spark.databricks.delta.retentionDurationCheck.enabled", "false")
            .config("spark.databricks.delta.merge.repartitionBeforeWrite", "true")
            .getOrCreate();
    }
    
    @Bean
    public DeltaTableManager deltaTableManager(SparkSession sparkSession) {
        return new DeltaTableManager(sparkSession);
    }
}

// Service de gestion des tables Delta avec optimisations
@Service
public class DeltaLakehouseService {
    
    private final DeltaTableManager deltaTableManager;
    private final DataQualityService dataQualityService;
    
    public void createOptimizedTable(String tableName, StructType schema) {
        // Création de la table avec optimisations Delta
        DeltaTable table = deltaTableManager.createTable(tableName, schema);
        
        // Configuration des optimisations
        table.optimize()
            .where("date >= '2024-01-01'")
            .executeCompaction();
        
        // Configuration de la rétention et des archives
        table.vacuum(168); // Garder 7 jours de données
    }
    
    public void upsertData(String tableName, Dataset<Row> newData, String mergeKey) {
        // Upsert avec gestion des conflits et audit trail
        DeltaTable table = deltaTableManager.getTable(tableName);
        
        table.alias("target")
            .merge(newData.alias("source"), mergeKey)
            .whenMatchedUpdateAll()
            .whenNotMatchedInsertAll()
            .execute();
        
        // Mise à jour des métriques de qualité
        dataQualityService.updateQualityMetrics(tableName);
    }
}