FinTech Neo-Bank Real-Time Pipeline

-- Table de faits principale pour les transactions
CREATE TABLE FACT_TRANSACTION (
    transaction_sk BIGINT PRIMARY KEY AUTO_INCREMENT,
    transaction_id VARCHAR(50) NOT NULL,
    transaction_timestamp TIMESTAMP(3) NOT NULL,
    sender_customer_sk INT,
    receiver_customer_sk INT,
    merchant_sk INT,
    date_sk INT NOT NULL,
    time_sk INT NOT NULL,
    transaction_type_sk INT NOT NULL,
    channel_sk INT NOT NULL,
    amount_euro DECIMAL(19,4) ENCRYPTED,
    fee_amount DECIMAL(19,4),
    fraud_score DECIMAL(3,2),
    is_flagged_fraud BOOLEAN DEFAULT FALSE,
    processing_time_ms INT,
    authorization_code VARCHAR(20),
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    
    INDEX idx_date_customer (date_sk, sender_customer_sk),
    INDEX idx_merchant_date (merchant_sk, date_sk),
    INDEX idx_fraud_score (fraud_score) WHERE fraud_score > 0.7
);

-- Table pour les décisions de crédit
CREATE TABLE FACT_CREDIT_DECISION (
    decision_sk BIGINT PRIMARY KEY AUTO_INCREMENT,
    customer_sk INT NOT NULL,
    application_id VARCHAR(50) NOT NULL,
    requested_amount DECIMAL(19,4),
    approved_amount DECIMAL(19,4),
    credit_score INT,
    risk_tier VARCHAR(20),
    decision_reason TEXT,
    decision_timestamp TIMESTAMP,
    
    INDEX idx_customer_decision (customer_sk, decision_timestamp)
);

-- Dimension client avec SCD Type 2
CREATE TABLE DIM_CUSTOMER (
    customer_sk INT PRIMARY KEY AUTO_INCREMENT,
    customer_id VARCHAR(50) NOT NULL,
    age_group VARCHAR(20),
    gender VARCHAR(10),
    location_tier VARCHAR(20),
    segment VARCHAR(20),
    risk_score VARCHAR(20),
    registration_date DATE,
    last_activity_date DATE,
    valid_from TIMESTAMP NOT NULL,
    valid_to TIMESTAMP NULL,
    is_current BOOLEAN DEFAULT TRUE,
    
    INDEX idx_customer_id (customer_id),
    INDEX idx_current (is_current)
);

-- Dimension marchand avec enrichissement des risques
CREATE TABLE DIM_MERCHANT (
    merchant_sk INT PRIMARY KEY AUTO_INCREMENT,
    merchant_id VARCHAR(50) NOT NULL,
    merchant_name VARCHAR(255),
    mcc_code VARCHAR(10),
    country_code VARCHAR(3),
    risk_level VARCHAR(20),
    fraud_rate DECIMAL(5,4),
    avg_transaction_amount DECIMAL(19,4),
    last_risk_assessment_date DATE,
    
    INDEX idx_merchant_id (merchant_id),
    INDEX idx_risk_level (risk_level)
);

-- Table de liaison pour la conformité PSD2
CREATE TABLE BRIDGE_TRANSACTION_COMPLIANCE (
    transaction_sk INT NOT NULL,
    regulation_type VARCHAR(20) NOT NULL,
    compliance_status VARCHAR(20),
    verification_timestamp TIMESTAMP,
    verification_method VARCHAR(50),
    compliance_score DECIMAL(3,2),
    
    PRIMARY KEY (transaction_sk, regulation_type),
    FOREIGN KEY (transaction_sk) REFERENCES FACT_TRANSACTION(transaction_sk)
);

-- Table pour le consentement GDPR
CREATE TABLE DIM_CONSENT (
    consent_sk INT PRIMARY KEY AUTO_INCREMENT,
    customer_sk INT NOT NULL,
    consent_type VARCHAR(50),
    consent_status VARCHAR(20),
    granted_date TIMESTAMP,
    revoked_date TIMESTAMP NULL,
    legal_basis VARCHAR(100),
    
    INDEX idx_customer_consent (customer_sk, consent_type),
    INDEX idx_status (consent_status)
);

-- Audit trail pour toutes les actions sensibles
CREATE TABLE FACT_AUDIT_TRAIL (
    audit_sk BIGINT PRIMARY KEY AUTO_INCREMENT,
    user_id VARCHAR(50),
    action_type VARCHAR(100),
    table_name VARCHAR(100),
    record_id VARCHAR(100),
    old_values JSON,
    new_values JSON,
    ip_address VARCHAR(45),
    user_agent TEXT,
    timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    
    INDEX idx_timestamp (timestamp),
    INDEX idx_action (action_type)
);

// Configuration Apache Kafka pour l'épine dorsale événementielle
@Configuration
public class KafkaConfig {
    
    @Bean
    public ProducerFactory<String, TransactionEvent> producerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        config.put(ProducerConfig.ACKS_CONFIG, "all");
        config.put(ProducerConfig.RETRIES_CONFIG, 3);
        config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        
        return new DefaultKafkaProducerFactory<>(config);
    }
    
    @Bean
    public ConsumerFactory<String, TransactionEvent> consumerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
        config.put(ConsumerConfig.GROUP_ID_CONFIG, "fraud-detection-group");
        config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        
        return new DefaultKafkaConsumerFactory<>(config);
    }
}

// Job Flink pour la détection de fraude en temps réel
public class FraudDetectionJob {
    
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(60000); // Checkpoint toutes les minutes
        env.setParallelism(4);
        
        // Source Kafka pour les transactions
        KafkaSource<TransactionEvent> source = KafkaSource.<TransactionEvent>builder()
            .setBootstrapServers("kafka:9092")
            .setTopics("transactions")
            .setGroupId("fraud-detection")
            .setStartingOffsets(OffsetsInitializer.latest())
            .setValueOnlyDeserializer(new TransactionEventDeserializer())
            .build();
        
        // Pipeline de détection de fraude
        env.fromSource(source, WatermarkStrategy.noWatermarks(), "Transaction Source")
            .keyBy(TransactionEvent::getCustomerId)
            .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
            .process(new FraudDetectionProcessor())
            .addSink(new FraudAlertSink());
        
        env.execute("Fraud Detection Job");
    }
}

// Processeur de détection de fraude
public class FraudDetectionProcessor extends KeyedProcessFunction<String, TransactionEvent, FraudAlert> {
    
    private ValueState<CustomerProfile> customerProfile;
    private ValueState<TransactionHistory> transactionHistory;
    
    @Override
    public void open(Configuration parameters) {
        customerProfile = getRuntimeContext().getState(
            new ValueStateDescriptor<>("customer-profile", CustomerProfile.class)
        );
        transactionHistory = getRuntimeContext().getState(
            new ValueStateDescriptor<>("transaction-history", TransactionHistory.class)
        );
    }
    
    @Override
    public void processElement(TransactionEvent event, Context ctx, Collector<FraudAlert> out) throws Exception {
        // Logique de détection de fraude
        double fraudScore = calculateFraudScore(event, customerProfile.value(), transactionHistory.value());
        
        if (fraudScore > 0.8) {
            FraudAlert alert = new FraudAlert(event.getTransactionId(), fraudScore, "High risk transaction");
            out.collect(alert);
        }
        
        // Mise à jour des états
        updateCustomerProfile(event);
        updateTransactionHistory(event);
    }
}

// Configuration DynamoDB pour les profils clients
@Configuration
public class DynamoDBConfig {
    
    @Bean
    public DynamoDbClient dynamoDbClient() {
        return DynamoDbClient.builder()
            .region(Region.EU_WEST_1)
            .credentialsProvider(DefaultCredentialsProvider.create())
            .build();
    }
    
    @Bean
    public DynamoDbEnhancedClient enhancedClient(DynamoDbClient client) {
        return DynamoDbEnhancedClient.builder()
            .dynamoDbClient(client)
            .build();
    }
}

// Repository pour les profils clients
@Repository
public class CustomerProfileRepository {
    
    private final DynamoDbEnhancedClient enhancedClient;
    private final DynamoDbTable<CustomerProfile> table;
    
    public CustomerProfileRepository(DynamoDbEnhancedClient enhancedClient) {
        this.enhancedClient = enhancedClient;
        this.table = enhancedClient.table("customer-profiles", 
            TableSchema.fromBean(CustomerProfile.class));
    }
    
    public CustomerProfile findByCustomerId(String customerId) {
        Key key = Key.builder().partitionValue(customerId).build();
        return table.getItem(key);
    }
    
    public void save(CustomerProfile profile) {
        table.putItem(profile);
    }
    
    // Requête avec index GSI pour les segments
    public List<CustomerProfile> findBySegment(String segment) {
        QueryEnhancedRequest request = QueryEnhancedRequest.builder()
            .queryConditional(QueryConditional.keyEqualTo(
                Key.builder().partitionValue("SEGMENT#" + segment).build()))
            .build();
        
        return table.index("segment-index").query(request)
            .stream()
            .map(Page::items)
            .flatMap(List::stream)
            .collect(Collectors.toList());
    }
}