Flink Streaming

+------------------+     CDC Events      +------------------+
| PostgreSQL       | ------------------> | Flink CDC        |
| (WAL Replication)|                     | Connector        |
+------------------+                     +--------+---------+
                                                  |
+------------------+     Kafka Events    +--------v---------+
| Kafka Topics     | ------------------> | Flink SQL        |
| - state-changes  |                     | Engine           |
| - agent-traces   |                     |                  |
| - llm-ops        |                     +--------+---------+
+------------------+                              |
                                         +--------v---------+
                                         | Iceberg Tables   |
                                         | (via Polaris)    |
                                         +------------------+

CREATE TEMPORARY TABLE fsm_transitions_cdc (
    id              STRING,
    entity_type     STRING NOT NULL,
    entity_id       STRING NOT NULL,
    tenant_id       STRING NOT NULL,
    from_state      STRING NOT NULL,
    to_state        STRING NOT NULL,
    triggered_by    STRING,
    trigger_reason  STRING,
    metadata        STRING,
    created_at      TIMESTAMP(3) NOT NULL,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'connector' = 'postgres-cdc',
    'hostname' = 'postgresql.matih-data-plane.svc.cluster.local',
    'port' = '5432',
    'username' = '${DATABASE_USER}',
    'password' = '${DATABASE_PASSWORD}',
    'database-name' = 'ai_service',
    'schema-name' = 'public',
    'table-name' = 'fsm_state_transitions',
    'slot.name' = 'flink_fsm_cdc',
    'decoding.plugin.name' = 'pgoutput'
);

INSERT INTO polaris.matih_analytics.state_transition_log
SELECT
    id              AS transition_id,
    entity_type,
    entity_id,
    tenant_id,
    from_state,
    to_state,
    triggered_by,
    trigger_reason,
    metadata,
    created_at      AS event_timestamp
FROM fsm_transitions_cdc;

export DATABASE_USER=ai_service
export DATABASE_PASSWORD=xxx
envsubst < state-transition-cdc.sql | curl -X POST sql-gateway/v1/statements

CREATE TEMPORARY TABLE state_changes_source (
    event_id        STRING,
    event_type      STRING,
    tenant_id       STRING NOT NULL,
    entity_type     STRING,
    entity_id       STRING,
    from_state      STRING,
    to_state        STRING,
    triggered_by    STRING,
    metadata        STRING,
    `timestamp`     TIMESTAMP(3),
    WATERMARK FOR `timestamp` AS `timestamp` - INTERVAL '30' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'matih.ai.state-changes',
    'properties.bootstrap.servers' = 'strimzi-kafka-kafka-bootstrap.matih-data-plane.svc.cluster.local:9093',
    'properties.group.id' = 'flink-session-analytics',
    'scan.startup.mode' = 'latest-offset',
    'format' = 'json'
);

INSERT INTO polaris.matih_analytics.session_analytics
SELECT
    TUMBLE_START(`timestamp`, INTERVAL '15' MINUTE) AS window_start,
    TUMBLE_END(`timestamp`, INTERVAL '15' MINUTE)   AS window_end,
    tenant_id,
    COUNT(DISTINCT entity_id)                         AS total_sessions,
    AVG(CAST(JSON_VALUE(metadata, '$.duration_ms') AS DOUBLE)) AS avg_duration_ms,
    AVG(CAST(JSON_VALUE(metadata, '$.turn_count') AS DOUBLE))  AS avg_turns,
    CAST(SUM(CASE WHEN to_state = 'completed' THEN 1 ELSE 0 END) AS DOUBLE)
        / GREATEST(COUNT(DISTINCT entity_id), 1)     AS completion_rate,
    COUNT(DISTINCT triggered_by)                      AS unique_users,
    TUMBLE_END(`timestamp`, INTERVAL '15' MINUTE)    AS event_timestamp
FROM state_changes_source
WHERE entity_type = 'session_phase'
GROUP BY
    TUMBLE(`timestamp`, INTERVAL '15' MINUTE),
    tenant_id;

CREATE TEMPORARY TABLE agent_traces_source (
    event_id        STRING,
    event_type      STRING,
    tenant_id       STRING NOT NULL,
    trace_id        STRING,
    agent_id        STRING NOT NULL,
    action          STRING,
    session_id      STRING,
    latency_ms      INT,
    tokens_input    INT,
    tokens_output   INT,
    cost_usd        DOUBLE,
    tools_used      ARRAY<STRING>,
    error_message   STRING,
    `timestamp`     TIMESTAMP(3),
    WATERMARK FOR `timestamp` AS `timestamp` - INTERVAL '30' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'matih.ai.agent-traces',
    'properties.bootstrap.servers' = 'strimzi-kafka-kafka-bootstrap.matih-data-plane.svc.cluster.local:9093',
    'properties.group.id' = 'flink-agent-perf-agg',
    'scan.startup.mode' = 'latest-offset',
    'format' = 'json'
);

INSERT INTO polaris.matih_analytics.agent_performance_metrics
SELECT
    TUMBLE_START(`timestamp`, INTERVAL '5' MINUTE)  AS window_start,
    TUMBLE_END(`timestamp`, INTERVAL '5' MINUTE)    AS window_end,
    tenant_id,
    agent_id,
    CAST(NULL AS STRING)                             AS agent_name,
    -- Success rate
    CAST(SUM(CASE WHEN action = 'completed' AND error_message IS NULL THEN 1 ELSE 0 END) AS DOUBLE)
        / GREATEST(COUNT(*), 1)                      AS success_rate,
    AVG(CAST(COALESCE(latency_ms, 0) AS DOUBLE))    AS avg_latency_ms,
    CAST(NULL AS BIGINT)                             AS p50_latency_ms,
    CAST(NULL AS BIGINT)                             AS p95_latency_ms,
    CAST(NULL AS BIGINT)                             AS p99_latency_ms,
    COUNT(*)                                          AS total_traces,
    SUM(CAST(COALESCE(tokens_input, 0) + COALESCE(tokens_output, 0) AS BIGINT)) AS total_tokens,
    SUM(COALESCE(cost_usd, 0.0))                     AS total_cost_usd,
    -- Error rate
    CAST(SUM(CASE WHEN error_message IS NOT NULL THEN 1 ELSE 0 END) AS DOUBLE)
        / GREATEST(COUNT(*), 1)                      AS error_rate,
    0.0                                               AS escalation_rate,
    TUMBLE_END(`timestamp`, INTERVAL '5' MINUTE)     AS event_timestamp
FROM agent_traces_source
WHERE action IN ('completed', 'failed')
GROUP BY
    TUMBLE(`timestamp`, INTERVAL '5' MINUTE),
    tenant_id,
    agent_id;

INSERT INTO polaris.matih_analytics.llm_operations_metrics
SELECT
    TUMBLE_START(`timestamp`, INTERVAL '15' MINUTE) AS window_start,
    TUMBLE_END(`timestamp`, INTERVAL '15' MINUTE)   AS window_end,
    tenant_id,
    COALESCE(provider, 'unknown')                    AS provider,
    COALESCE(model, 'unknown')                       AS model,
    -- Cache hit rate
    CAST(SUM(CASE WHEN cache_hit = TRUE THEN 1 ELSE 0 END) AS DOUBLE)
        / GREATEST(COUNT(*), 1)                      AS cache_hit_rate,
    AVG(CAST(COALESCE(tokens_input, 0) AS DOUBLE))  AS avg_tokens_input,
    AVG(CAST(COALESCE(tokens_output, 0) AS DOUBLE)) AS avg_tokens_output,
    COUNT(*)                                          AS total_requests,
    SUM(COALESCE(cost_usd, 0.0))                     AS total_cost_usd,
    AVG(CAST(COALESCE(latency_ms, 0) AS DOUBLE))    AS avg_latency_ms,
    -- Error rate
    CAST(SUM(CASE WHEN success = FALSE THEN 1 ELSE 0 END) AS DOUBLE)
        / GREATEST(COUNT(*), 1)                      AS error_rate,
    TUMBLE_END(`timestamp`, INTERVAL '15' MINUTE)    AS event_timestamp
FROM llm_ops_source
GROUP BY
    TUMBLE(`timestamp`, INTERVAL '15' MINUTE),
    tenant_id,
    provider,
    model;

class FlinkJobService:
    """Manages Flink SQL jobs."""
 
    async def submit_sql_job(self, sql: str, job_name: str) -> FlinkJob:
        """Submit a Flink SQL job to the SQL Gateway."""
 
    async def get_job_status(self, job_id: str) -> FlinkJobStatus:
        """Get the status of a running Flink job."""
 
    async def cancel_job(self, job_id: str) -> bool:
        """Cancel a running Flink job."""
 
    async def create_savepoint(self, job_id: str, path: str) -> str:
        """Create a savepoint for a Flink job."""
 
    async def resume_from_savepoint(self, job_id: str, savepoint_path: str) -> FlinkJob:
        """Resume a job from a savepoint."""

class FlinkSavepointService:
    """Manages Flink savepoints for graceful upgrades."""
 
    async def trigger_savepoint(self, job_id: str) -> Savepoint:
        # 1. Trigger savepoint on running job
        # 2. Wait for savepoint completion
        # 3. Store savepoint path in metadata
 
    async def restore_from_savepoint(self, job_id: str, savepoint: Savepoint) -> FlinkJob:
        # 1. Cancel current job (if running)
        # 2. Submit new job with savepoint path
        # 3. Verify job is consuming from savepoint offset

WATERMARK FOR `timestamp` AS `timestamp` - INTERVAL '30' SECOND