-- Migration 0031: Phase 2-3 Observability & API Pooling Infrastructure -- Purpose: Add tables for OpenDart caching, KIS pool, rate limiting, circuit breaker -- ============================================================================ -- OPENDATA SCHEMA: OpenDart Financial Data Caching -- ============================================================================ CREATE SCHEMA IF NOT EXISTS opendata; -- OpenDart cache (quarterly financials) CREATE TABLE IF NOT EXISTS opendata.opendart_cache ( id BIGSERIAL PRIMARY KEY, ticker VARCHAR(10) NOT NULL, quarter VARCHAR(6) NOT NULL, -- YYYY-QN format data_json JSONB NOT NULL, cached_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, expires_at TIMESTAMP WITH TIME ZONE NOT NULL, published_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, -- Unique constraint: one cache entry per ticker/quarter UNIQUE(ticker, quarter) ); CREATE INDEX IF NOT EXISTS idx_opendart_cache_ticker ON opendata.opendart_cache(ticker); CREATE INDEX IF NOT EXISTS idx_opendart_cache_expires_at ON opendata.opendart_cache(expires_at); CREATE INDEX IF NOT EXISTS idx_opendart_cache_published_at ON opendata.opendart_cache(published_at); -- OpenDart batch execution log CREATE TABLE IF NOT EXISTS opendata.opendart_batch_log ( id BIGSERIAL PRIMARY KEY, batch_date DATE NOT NULL, quota_limit INT NOT NULL DEFAULT 1000, quota_used INT NOT NULL DEFAULT 0, status VARCHAR(50) NOT NULL, -- 'success', 'quota_exceeded', 'partial', 'failed' error_message TEXT, executed_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, published_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, -- Unique constraint: one batch per day UNIQUE(batch_date) ); CREATE INDEX IF NOT EXISTS idx_opendart_batch_log_batch_date ON opendata.opendart_batch_log(batch_date); CREATE INDEX IF NOT EXISTS idx_opendart_batch_log_status ON opendata.opendart_batch_log(status); -- ============================================================================ -- KIS SCHEMA: Korea Investment & Securities Connection Pool -- ============================================================================ CREATE SCHEMA IF NOT EXISTS kis; -- KIS connection pool state CREATE TABLE IF NOT EXISTS kis.connection_pool_state ( id BIGSERIAL PRIMARY KEY, connection_id UUID NOT NULL, state VARCHAR(50) NOT NULL, -- 'idle', 'active', 'closed' priority INT NOT NULL, -- 0=BUY, 1=SELL, 2=CANCEL token_hash VARCHAR(256), -- Hash of OAuth2 token (PII protection) expires_at TIMESTAMP WITH TIME ZONE, created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, released_at TIMESTAMP WITH TIME ZONE, published_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, -- Unique constraint: one state per connection_id UNIQUE(connection_id) ); CREATE INDEX IF NOT EXISTS idx_kis_connection_pool_state ON kis.connection_pool_state(state); CREATE INDEX IF NOT EXISTS idx_kis_connection_pool_expires_at ON kis.connection_pool_state(expires_at); CREATE INDEX IF NOT EXISTS idx_kis_connection_pool_priority ON kis.connection_pool_state(priority); -- KIS token refresh log CREATE TABLE IF NOT EXISTS kis.token_refresh_log ( id BIGSERIAL PRIMARY KEY, connection_id UUID NOT NULL, refresh_at TIMESTAMP WITH TIME ZONE NOT NULL, status VARCHAR(50) NOT NULL, -- 'success', 'failed', 'expired' error_message TEXT, new_token_hash VARCHAR(256), executed_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, published_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP ); CREATE INDEX IF NOT EXISTS idx_kis_token_refresh_connection_id ON kis.token_refresh_log(connection_id); CREATE INDEX IF NOT EXISTS idx_kis_token_refresh_status ON kis.token_refresh_log(status); CREATE INDEX IF NOT EXISTS idx_kis_token_refresh_executed_at ON kis.token_refresh_log(executed_at); -- ============================================================================ -- INFRASTRUCTURE SCHEMA: Rate Limiting & Circuit Breaker -- ============================================================================ CREATE SCHEMA IF NOT EXISTS infrastructure; -- Rate limit quota tracking (per API) CREATE TABLE IF NOT EXISTS infrastructure.rate_limit_quota ( id BIGSERIAL PRIMARY KEY, api_name VARCHAR(50) NOT NULL, -- 'krx', 'opendart', 'kis' limit_count INT NOT NULL, -- e.g., 100 for KRX window_seconds INT NOT NULL, -- e.g., 60 for per-minute current_tokens DECIMAL(10, 2) NOT NULL DEFAULT 0, last_reset_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, published_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, -- Unique constraint: one quota per API UNIQUE(api_name) ); CREATE INDEX IF NOT EXISTS idx_rate_limit_quota_api_name ON infrastructure.rate_limit_quota(api_name); -- Rate limit events (for audit trail) CREATE TABLE IF NOT EXISTS infrastructure.rate_limit_events ( id BIGSERIAL PRIMARY KEY, api_name VARCHAR(50) NOT NULL, request_id UUID, decision VARCHAR(50) NOT NULL, -- 'allowed', 'rejected' tokens_requested INT NOT NULL, tokens_used INT NOT NULL, remaining_tokens DECIMAL(10, 2) NOT NULL, occurred_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, published_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP ); CREATE INDEX IF NOT EXISTS idx_rate_limit_events_api_name ON infrastructure.rate_limit_events(api_name); CREATE INDEX IF NOT EXISTS idx_rate_limit_events_occurred_at ON infrastructure.rate_limit_events(occurred_at); -- Circuit breaker state CREATE TABLE IF NOT EXISTS infrastructure.circuit_breaker_state ( id BIGSERIAL PRIMARY KEY, api_name VARCHAR(50) NOT NULL, state VARCHAR(50) NOT NULL, -- 'closed', 'open', 'half_open' consecutive_errors INT NOT NULL DEFAULT 0, last_error_at TIMESTAMP WITH TIME ZONE, opened_at TIMESTAMP WITH TIME ZONE, closed_at TIMESTAMP WITH TIME ZONE, updated_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, published_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, -- Unique constraint: one state per API UNIQUE(api_name) ); CREATE INDEX IF NOT EXISTS idx_circuit_breaker_state_api_name ON infrastructure.circuit_breaker_state(api_name); -- Circuit breaker events (for audit trail) CREATE TABLE IF NOT EXISTS infrastructure.circuit_breaker_events ( id BIGSERIAL PRIMARY KEY, api_name VARCHAR(50) NOT NULL, state_transition VARCHAR(50) NOT NULL, -- e.g., 'closed→open', 'open→half_open' error_count INT NOT NULL, error_message TEXT, occurred_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, published_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP ); CREATE INDEX IF NOT EXISTS idx_circuit_breaker_events_api_name ON infrastructure.circuit_breaker_events(api_name); CREATE INDEX IF NOT EXISTS idx_circuit_breaker_events_occurred_at ON infrastructure.circuit_breaker_events(occurred_at); -- ============================================================================ -- OBSERVABILITY SCHEMA: Metrics & Monitoring -- ============================================================================ CREATE SCHEMA IF NOT EXISTS observability; -- Batch SLA metrics CREATE TABLE IF NOT EXISTS observability.batch_sla_metrics ( id BIGSERIAL PRIMARY KEY, job_name VARCHAR(100) NOT NULL, job_type VARCHAR(50) NOT NULL, -- 'recommendation_report', 'opendart_batch', etc. started_at TIMESTAMP WITH TIME ZONE NOT NULL, completed_at TIMESTAMP WITH TIME ZONE NOT NULL, duration_seconds INT NOT NULL, status VARCHAR(50) NOT NULL, -- 'success', 'failed', 'timeout' recorded_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, published_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP ); CREATE INDEX IF NOT EXISTS idx_batch_sla_job_name ON observability.batch_sla_metrics(job_name); CREATE INDEX IF NOT EXISTS idx_batch_sla_completed_at ON observability.batch_sla_metrics(completed_at); -- Data quality quarantine (rows marked for manual review) CREATE TABLE IF NOT EXISTS observability.data_quality_quarantine ( id BIGSERIAL PRIMARY KEY, module_name VARCHAR(100) NOT NULL, reason VARCHAR(256) NOT NULL, -- e.g., 'missing_required_field', 'invalid_state_transition' entity_id UUID, entity_type VARCHAR(50), quarantined_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, resolution_status VARCHAR(50), -- NULL, 'resolved', 'ignored' published_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP ); CREATE INDEX IF NOT EXISTS idx_data_quality_module ON observability.data_quality_quarantine(module_name); CREATE INDEX IF NOT EXISTS idx_data_quality_quarantined_at ON observability.data_quality_quarantine(quarantined_at); -- ============================================================================ -- APPEND-ONLY AUDIT TRAIL (for all Phase 2-3 operations) -- ============================================================================ CREATE TABLE IF NOT EXISTS infrastructure.operation_audit_trail ( id BIGSERIAL PRIMARY KEY, operation_type VARCHAR(50) NOT NULL, -- 'opendata_batch', 'kis_token_refresh', 'rate_limit_check', etc. operation_id UUID NOT NULL, correlation_id UUID, status VARCHAR(50) NOT NULL, -- 'initiated', 'in_progress', 'completed', 'failed' metadata_json JSONB, occurred_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, published_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP ); CREATE INDEX IF NOT EXISTS idx_operation_audit_type ON infrastructure.operation_audit_trail(operation_type); CREATE INDEX IF NOT EXISTS idx_operation_audit_correlation_id ON infrastructure.operation_audit_trail(correlation_id); CREATE INDEX IF NOT EXISTS idx_operation_audit_occurred_at ON infrastructure.operation_audit_trail(occurred_at); -- Permissions: schemas owned by executing role; no explicit role-based GRANT in dev/test -- In production, add explicit role-based GRANT via separate admin script after schema creation