starting dynamodb -> mysql/mariadb migration

Author: emidiocruciani

cmd/main/config/config.yaml

@@ -16,12 +16,25 @@ health-check:
   server:
     port: 8080
 
+mysql:
+  host: "${MYSQL_HOST}"
+  port: ${MYSQL_PORT}
+  user: "${MYSQL_USER}"
+  password: "${MYSQL_PASSWORD}"
+  database: "${MYSQL_DATABASE}"
+
 outbox:
   table-name: "${EMAIL_OUTBOX_TABLE}"
 
 pipeline:
   interval: ${PIPELINE_INTERVAL}
 
+pipelines:
+  dynamodb:
+    enabled: ${DYNAMODB_PIPELINES_ENABLED}
+  mysql:
+    enabled: false
+
 smtp:
   host: "${SMTP_HOST}"
   port: ${SMTP_PORT}

compose.yml

@@ -25,6 +25,13 @@ services:
       ATTACHMENTS_BASE_PATH: 'testdata/attachments'
       EMAIL_OUTBOX_TABLE: 'Outbox'
       EML_STORAGE_PATH: 'testdata/.out/eml'
+      MYSQL_HOST: '127.0.0.1'
+      MYSQL_PORT: '3306'
+      MYSQL_USER: 'root'
+      MYSQL_PASSWORD: 'test'
+      MYSQL_DATABASE: 'mailculator_test'
+      DYNAMODB_PIPELINES_ENABLED: 'true'
+      MYSQL_PIPELINES_ENABLED: 'true'
     command: >
       sh -c "go mod tidy &&
              ./scripts/coverage.sh unit &&
@@ -153,3 +160,64 @@ services:
       AWS_SECRET_ACCESS_KEY: 'local'
     networks:
       - mailculator-processor-deployments-net
+
+  mysql: &mysql-base
+    profiles:
+      - 'none'
+    image: mariadb:10.11
+    environment:
+      MARIADB_ROOT_PASSWORD: 'test'
+      MARIADB_DATABASE: 'mailculator_test'
+    volumes:
+      - './docker/mysql/migrations:/docker-entrypoint-initdb.d/migrations:ro'
+      - './docker/mysql/init.sh:/docker-entrypoint-initdb.d/zzz-init.sh:ro'
+    healthcheck:
+      test: ['CMD-SHELL', 'test -f /var/lib/mysql/zz-finish && mysqladmin ping -h localhost -ptest']
+      interval: 2s
+      timeout: 5s
+      retries: 30
+    ports:
+      - '127.0.0.1:3306:3306'
+    networks:
+      - mailculator-processor-deployments-net
+
+  mysql-test:
+    <<: *mysql-base
+    container_name: mailculator_processor_mysql_test
+    profiles:
+      - 'test-deps'
+
+  mysql-devcontainer:
+    <<: *mysql-base
+    container_name: mailculator_processor_mysql_devcontainer
+    profiles:
+      - 'devcontainer-deps'
+
+  wait-for-mysql-test:
+    container_name: mailculator_processor_wait_for_mysql_test
+    profiles:
+      - 'test-deps'
+    image: golang:1.25-alpine
+    command: ['echo', 'Service mysql-test is ready']
+    networks:
+      - mailculator-processor-deployments-net
+    depends_on:
+      mysql-test:
+        condition: service_healthy
+
+  phpmyadmin:
+    container_name: mailculator_processor_phpmyadmin_devcontainer
+    profiles:
+      - 'devcontainer-deps'
+    image: phpmyadmin:latest
+    ports:
+      - '127.0.0.1:9003:80'
+    environment:
+      PMA_HOST: 'mysql-devcontainer'
+      PMA_USER: 'root'
+      PMA_PASSWORD: 'test'
+    networks:
+      - mailculator-processor-deployments-net
+    depends_on:
+      mysql-devcontainer:
+        condition: service_healthy

docker/mysql/init.sh

@@ -0,0 +1,14 @@
+#!/bin/bash
+set -e
+
+echo "Running MariaDB migrations..."
+
+for f in /docker-entrypoint-initdb.d/migrations/*.sql; do
+    echo "Executing $f..."
+    mysql -u root -p"$MARIADB_ROOT_PASSWORD" "$MARIADB_DATABASE" < "$f"
+done
+
+echo "Migrations completed. Creating finish marker..."
+touch /var/lib/mysql/zz-finish
+
+echo "MariaDB initialization complete."

docker/mysql/migrations/001_create_emails_table.sql

@@ -0,0 +1,18 @@
+CREATE TABLE IF NOT EXISTS emails (
+    id CHAR(36) PRIMARY KEY,
+    status ENUM(
+        'ACCEPTED','INTAKING','READY','PROCESSING',
+        'SENT','FAILED','INVALID',
+        'CALLING-SENT-CALLBACK','CALLING-FAILED-CALLBACK',
+        'SENT-ACKNOWLEDGED','FAILED-ACKNOWLEDGED'
+    ) NOT NULL,
+    eml_file_path VARCHAR(500),
+    payload_file_path VARCHAR(500),
+    reason TEXT,
+    version INT NOT NULL DEFAULT 1,
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+
+    INDEX idx_status (status),
+    INDEX idx_status_updated (status, updated_at)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

docker/mysql/migrations/002_create_email_statuses_table.sql

@@ -0,0 +1,10 @@
+CREATE TABLE IF NOT EXISTS email_statuses (
+    id BIGINT AUTO_INCREMENT PRIMARY KEY,
+    email_id CHAR(36) NOT NULL,
+    status VARCHAR(50) NOT NULL,
+    reason TEXT,
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+
+    INDEX idx_email_id (email_id),
+    FOREIGN KEY (email_id) REFERENCES emails(id) ON DELETE CASCADE
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

docs/database.md

@@ -17,6 +17,7 @@ type Email struct {
     UpdatedAt       string  // Timestamp RFC3339 dell'ultimo aggiornamento
     Reason          string  // Motivo dell'ultimo stato
     TTL             *int64  // Timestamp Unix in secondi per DynamoDB TTL (nil se non presente)
+    Version         int     // Versione per optimistic locking (MySQL) o 0 (DynamoDB)
 }
 ```
 
@@ -34,7 +35,7 @@ DynamoDB TTL è configurato per eliminare automaticamente i record obsoleti. L'a
 ttl := time.Now().Add(7 * 24 * time.Hour).Unix()
 ```
 
-## Pattern di Versionamento
+## Pattern di Versionamento (DynamoDB)
 
 ### Status Meta
 - **Costante**: `StatusMeta = "_META"`
@@ -63,6 +64,101 @@ INSERT INTO "table" VALUE {'Id': ?, 'Status': ?, 'Attributes': ?, 'TTL': ?}
 
 **Nota**: Il TTL viene sempre sincronizzato tra il record _META e i record di stato. Quando un TTL è presente, viene impostato sia alla radice del record _META che alla radice del nuovo record di stato.
 
+## Unificazione dei Tipi
+
+Entrambi i backend (DynamoDB e MySQL) utilizzano ora lo stesso tipo `Email` con tutti i campi necessari:
+
+- **DynamoDB**: `Version = 0` (non usa locking basato su versione)
+- **MySQL**: `Version` popolato dal database per optimistic locking
+- **TTL**: Presente per DynamoDB, `nil` per MySQL (non supportato)
+
+Questo approccio elimina la duplicazione dei tipi e semplifica l'architettura.
+
+---
+
+## MySQL Schema
+
+### Tabella `emails`
+Tabella principale per la gestione delle email.
+
+```sql
+CREATE TABLE IF NOT EXISTS emails (
+    id CHAR(36) PRIMARY KEY,
+    status ENUM(
+        'ACCEPTED','INTAKING','READY','PROCESSING',
+        'SENT','FAILED','INVALID',
+        'CALLING-SENT-CALLBACK','CALLING-FAILED-CALLBACK',
+        'SENT-ACKNOWLEDGED','FAILED-ACKNOWLEDGED'
+    ) NOT NULL,
+    eml_file_path VARCHAR(500),
+    payload_file_path VARCHAR(500),
+    reason TEXT,
+    version INT NOT NULL DEFAULT 1,
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+
+    INDEX idx_status (status),
+    INDEX idx_status_updated (status, updated_at)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+```
+
+**Nota**: A differenza di DynamoDB, MySQL non supporta TTL nativo. La pulizia dei record obsoleti deve essere gestita esternamente.
+
+### Tabella `email_statuses`
+Tabella per lo storico dei cambi di stato (history).
+
+```sql
+CREATE TABLE IF NOT EXISTS email_statuses (
+    id BIGINT AUTO_INCREMENT PRIMARY KEY,
+    email_id CHAR(36) NOT NULL,
+    status VARCHAR(50) NOT NULL,
+    reason TEXT,
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+
+    INDEX idx_email_id (email_id),
+    FOREIGN KEY (email_id) REFERENCES emails(id) ON DELETE CASCADE
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+```
+
+### Optimistic Locking (MySQL)
+MySQL utilizza optimistic locking basato su:
+- Campo `Version` nel tipo `Email` per tracciare le modifiche
+- Campo `status` per validare la transizione di stato
+
+Ogni update incrementa la versione e verifica lo stato atteso:
+```sql
+UPDATE emails
+SET status = ?, reason = ?, version = version + 1
+WHERE id = ? AND status = ?
+```
+
+Se `affected_rows = 0`, l'operazione restituisce `ErrLockNotAcquired`.
+
+**Nota**: DynamoDB non usa version-based locking, quindi restituisce sempre `Version = 0`.
+
+### Query con SKIP LOCKED
+Le query di lettura utilizzano `FOR UPDATE SKIP LOCKED` per:
+- Evitare blocchi su righe già in uso da altri worker
+- Migliorare il throughput in scenari con più processori concorrenti
+
+```sql
+SELECT id, status, eml_file_path, payload_file_path, reason, version, updated_at
+FROM emails
+WHERE status = ?
+ORDER BY updated_at ASC
+LIMIT ?
+FOR UPDATE SKIP LOCKED
+```
+
+### Transazioni
+Le operazioni di update e insert history sono eseguite in transazione per garantire atomicità:
+1. `BEGIN`
+2. `UPDATE emails ...`
+3. `INSERT INTO email_statuses ...`
+4. `COMMIT` (o `ROLLBACK` in caso di errore)
+
+---
+
 ## Stati Disponibili
 - `ACCEPTED` - Email accettato, in attesa di intake
 - `INTAKING` - Email in fase di elaborazione intake
@@ -76,7 +172,7 @@ INSERT INTO "table" VALUE {'Id': ?, 'Status': ?, 'Attributes': ?, 'TTL': ?}
 - `SENT-ACKNOWLEDGED` - Callback per email inviato completato
 - `FAILED-ACKNOWLEDGED` - Callback per email fallito completato
 
-## Query Pattern
+## Query Pattern (DynamoDB)
 
 ### Query per Stato
 ```sql
@@ -90,7 +186,7 @@ WHERE Status=? AND Attributes.Latest =?
 - Utilizza `NextToken` di DynamoDB per paginazione automatica
 - Interrompe quando raggiunge il limite di 25 record
 
-## PartiQL Operations
+## PartiQL Operations (DynamoDB)
 
 ### ExecuteStatement
 Utilizzato per query con parametri e paginazione.

docs/error-handling.md

@@ -19,6 +19,32 @@ Il sistema effettua retry automatico per le seguenti eccezioni DynamoDB:
 - **Formula**: Durata casuale tra 0 e min(2^attempt * base_delay, max_delay)
 
 
+## Retry MySQL
+
+### Condizioni di Retry
+Il sistema effettua retry automatico per i seguenti errori MySQL:
+- `1205` - Lock wait timeout exceeded
+- `1213` - Deadlock found when trying to get lock
+- `1040` - Too many connections
+- `1203` - User already has more than max_user_connections active connections
+- `driver.ErrBadConn` - Connessione al database persa
+
+### Errori NON Soggetti a Retry
+- `ErrLockNotAcquired` - Conflitto di lock ottimistico (il record è stato modificato da un altro processo)
+
+### Backoff Strategy
+- **Max Attempts**: 8 tentativi
+- **Base Delay**: 30 millisecondi
+- **Max Delay**: 1 secondo
+- **Formula**: Durata casuale tra 0 e min(2^attempt * base_delay, max_delay)
+
+### Transazioni
+Le operazioni MySQL (Update, Ready, Create) sono eseguite in transazione:
+- In caso di errore, viene eseguito automaticamente il rollback
+- In caso di successo, viene eseguito il commit
+- Gli errori transitori sono gestiti con retry (la transazione viene ritentata dall'inizio)
+
+
 ## Retry Callback HTTP
 
 ### Condizioni di Retry
@@ -43,6 +69,7 @@ Quando il callback HTTP fallisce dopo tutti i retry:
 Quando non riesce ad acquisire il lock di processamento:
 - Operazione saltata
 - Log warning: "failed to acquire processing lock"
+- **Nessun retry**: `ErrLockNotAcquired` indica che un altro worker sta già processando il record
 
 ## Context Cancellation
 Tutti i retry rispettano il context cancellation: