← back to Watches
database/enhanced-schema.sql
777 lines
-- ============================================================================
-- OMEGA WATCH PRICE HISTORY - ENTERPRISE POSTGRESQL SCHEMA
-- ============================================================================
-- Author: Database Architect
-- Version: 2.0 - Enterprise Edition
-- Features:
-- - Read replicas support
-- - Connection pooling ready
-- - Materialized views for analytics
-- - Full-text search with GIN indices
-- - Table partitioning for scalability
-- - Comprehensive indexing strategy
-- - ETL pipeline support
-- - Audit logging
-- - Performance optimization
-- ============================================================================
-- Drop existing objects if re-running (development only)
DROP SCHEMA IF EXISTS omega CASCADE;
CREATE SCHEMA omega;
SET search_path TO omega, public;
-- Enable required extensions
CREATE EXTENSION IF NOT EXISTS "uuid-ossp";
CREATE EXTENSION IF NOT EXISTS "pg_trgm"; -- Trigram matching for fuzzy search
CREATE EXTENSION IF NOT EXISTS "btree_gin"; -- GIN indexing for btree types
CREATE EXTENSION IF NOT EXISTS "pg_stat_statements"; -- Query performance tracking
-- ============================================================================
-- CUSTOM DOMAIN TYPES & ENUMS
-- ============================================================================
CREATE DOMAIN positive_price AS NUMERIC(12,2)
CHECK (VALUE >= 0);
CREATE DOMAIN positive_integer AS INTEGER
CHECK (VALUE >= 0);
CREATE DOMAIN watch_year AS INTEGER
CHECK (VALUE >= 1848 AND VALUE <= EXTRACT(YEAR FROM CURRENT_DATE) + 5);
CREATE TYPE watch_series_enum AS ENUM (
'Speedmaster', 'Seamaster', 'Constellation', 'De Ville',
'Railmaster', 'Flightmaster', 'Cosmic', 'Genève', 'Specialty'
);
CREATE TYPE condition_enum AS ENUM (
'new', 'excellent', 'good', 'fair', 'vintage', 'prototype'
);
CREATE TYPE case_material_enum AS ENUM (
'stainless_steel', 'gold_18k', 'gold_14k', 'platinum',
'titanium', 'ceramic', 'bronze', 'two_tone',
'canopus_gold', 'sedna_gold', 'moonshine_gold'
);
CREATE TYPE movement_type_enum AS ENUM (
'automatic', 'manual', 'quartz', 'spring_drive', 'co_axial'
);
-- ============================================================================
-- CORE TABLES WITH OPTIMIZED STRUCTURE
-- ============================================================================
-- Watches: Main entity with comprehensive metadata
CREATE TABLE watches (
id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
slug VARCHAR(255) UNIQUE NOT NULL,
model VARCHAR(255) NOT NULL,
series watch_series_enum NOT NULL,
reference VARCHAR(100) NOT NULL,
year_introduced watch_year NOT NULL,
description TEXT,
detailed_description TEXT,
image_url VARCHAR(500),
-- Limited edition tracking
is_limited_edition BOOLEAN DEFAULT false,
limited_edition_count INTEGER,
production_years INT4RANGE, -- Range type for production period
-- Metadata
created_at TIMESTAMPTZ DEFAULT NOW(),
updated_at TIMESTAMPTZ DEFAULT NOW(),
is_active BOOLEAN DEFAULT true,
is_featured BOOLEAN DEFAULT false,
-- Full-text search vector (auto-updated via trigger)
search_vector tsvector,
-- Data quality
data_quality_score INTEGER CHECK (data_quality_score >= 0 AND data_quality_score <= 100),
-- Constraints
CONSTRAINT valid_slug CHECK (slug ~* '^[a-z0-9-]+$'),
CONSTRAINT valid_limited_edition CHECK (
(is_limited_edition = false AND limited_edition_count IS NULL) OR
(is_limited_edition = true AND limited_edition_count > 0)
)
);
-- Comprehensive indexing strategy for watches
CREATE INDEX idx_watches_series ON watches(series) WHERE is_active = true;
CREATE INDEX idx_watches_year ON watches(year_introduced);
CREATE INDEX idx_watches_year_series ON watches(year_introduced, series);
CREATE INDEX idx_watches_active ON watches(is_active) WHERE is_active = true;
CREATE INDEX idx_watches_featured ON watches(is_featured) WHERE is_featured = true;
CREATE INDEX idx_watches_search ON watches USING GIN(search_vector);
CREATE INDEX idx_watches_model_trgm ON watches USING GIN(model gin_trgm_ops);
CREATE INDEX idx_watches_reference ON watches(reference);
-- Specifications: Technical details with normalized structure
CREATE TABLE specifications (
id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
watch_id UUID NOT NULL UNIQUE REFERENCES watches(id) ON DELETE CASCADE,
-- Case specifications
case_size VARCHAR(50),
case_diameter NUMERIC(5,2), -- Extracted numeric value for queries
case_material case_material_enum,
case_thickness VARCHAR(50),
case_thickness_mm NUMERIC(5,2),
lug_width VARCHAR(50),
lug_width_mm NUMERIC(5,2),
-- Movement specifications
movement_calibre VARCHAR(100),
movement_type movement_type_enum,
power_reserve VARCHAR(50),
power_reserve_hours INTEGER,
frequency_vph INTEGER, -- Vibrations per hour
jewel_count INTEGER,
-- Water resistance
water_resistance VARCHAR(50),
water_resistance_meters INTEGER,
-- Crystal and aesthetics
crystal_type VARCHAR(50),
dial_color VARCHAR(100),
bezel_type VARCHAR(100),
bracelet_type VARCHAR(100),
-- Physical properties
weight VARCHAR(50),
weight_grams INTEGER,
-- Additional metadata
created_at TIMESTAMPTZ DEFAULT NOW(),
updated_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX idx_specifications_watch_id ON specifications(watch_id);
CREATE INDEX idx_specifications_material ON specifications(case_material);
CREATE INDEX idx_specifications_movement ON specifications(movement_type);
CREATE INDEX idx_specifications_diameter ON specifications(case_diameter) WHERE case_diameter IS NOT NULL;
CREATE INDEX idx_specifications_calibre ON specifications(movement_calibre);
-- Features: Normalized feature taxonomy
CREATE TABLE features (
id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
name VARCHAR(100) UNIQUE NOT NULL,
category VARCHAR(50) NOT NULL, -- 'complication', 'certification', 'functionality'
description TEXT,
icon_name VARCHAR(50),
created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX idx_features_category ON features(category);
CREATE INDEX idx_features_name ON features(name);
-- Watch Features: Junction table with metadata
CREATE TABLE watch_features (
watch_id UUID REFERENCES watches(id) ON DELETE CASCADE,
feature_id UUID REFERENCES features(id) ON DELETE CASCADE,
display_order INTEGER DEFAULT 0,
created_at TIMESTAMPTZ DEFAULT NOW(),
PRIMARY KEY (watch_id, feature_id)
);
CREATE INDEX idx_watch_features_watch ON watch_features(watch_id);
CREATE INDEX idx_watch_features_feature ON watch_features(feature_id);
-- ============================================================================
-- PRICE HISTORY - PARTITIONED FOR SCALABILITY
-- ============================================================================
CREATE TABLE price_history (
id UUID DEFAULT uuid_generate_v4(),
watch_id UUID NOT NULL REFERENCES watches(id) ON DELETE CASCADE,
year watch_year NOT NULL,
price positive_price NOT NULL,
condition condition_enum NOT NULL DEFAULT 'new',
source VARCHAR(255),
notes TEXT,
currency CHAR(3) DEFAULT 'USD',
-- Auction/Sale metadata
auction_house VARCHAR(100),
lot_number VARCHAR(50),
sale_date DATE,
buyer_premium_pct NUMERIC(5,2),
-- Data quality indicators
is_estimate BOOLEAN DEFAULT false,
confidence_level INTEGER CHECK (confidence_level >= 1 AND confidence_level <= 5),
verified BOOLEAN DEFAULT false,
-- Price adjustments
inflation_adjusted_price NUMERIC(12,2),
exchange_rate NUMERIC(10,6),
-- Metadata
created_at TIMESTAMPTZ DEFAULT NOW(),
updated_at TIMESTAMPTZ DEFAULT NOW(),
PRIMARY KEY (id, year)
) PARTITION BY RANGE (year);
-- Create partitions by decade for optimal query performance
CREATE TABLE price_history_pre_1960 PARTITION OF price_history
FOR VALUES FROM (1848) TO (1960);
CREATE TABLE price_history_1960s PARTITION OF price_history
FOR VALUES FROM (1960) TO (1970);
CREATE TABLE price_history_1970s PARTITION OF price_history
FOR VALUES FROM (1970) TO (1980);
CREATE TABLE price_history_1980s PARTITION OF price_history
FOR VALUES FROM (1980) TO (1990);
CREATE TABLE price_history_1990s PARTITION OF price_history
FOR VALUES FROM (1990) TO (2000);
CREATE TABLE price_history_2000s PARTITION OF price_history
FOR VALUES FROM (2000) TO (2010);
CREATE TABLE price_history_2010s PARTITION OF price_history
FOR VALUES FROM (2010) TO (2020);
CREATE TABLE price_history_2020s PARTITION OF price_history
FOR VALUES FROM (2020) TO (2030);
CREATE TABLE price_history_future PARTITION OF price_history
FOR VALUES FROM (2030) TO (2050);
-- Comprehensive indexing on partitioned table
CREATE INDEX idx_price_history_watch_id ON price_history(watch_id);
CREATE INDEX idx_price_history_watch_year ON price_history(watch_id, year);
CREATE INDEX idx_price_history_year_price ON price_history(year, price);
CREATE INDEX idx_price_history_condition ON price_history(condition);
CREATE INDEX idx_price_history_verified ON price_history(verified) WHERE verified = true;
-- ============================================================================
-- MATERIALIZED VIEWS FOR ANALYTICS & PERFORMANCE
-- ============================================================================
-- Watch Statistics: Pre-calculated metrics for fast queries
CREATE MATERIALIZED VIEW watch_statistics AS
SELECT
w.id,
w.slug,
w.model,
w.series,
w.reference,
w.year_introduced,
-- Price point metrics
COUNT(DISTINCT ph.id) as price_point_count,
MIN(ph.year) as earliest_price_year,
MAX(ph.year) as latest_price_year,
-- Price statistics
MIN(ph.price) as lowest_price_ever,
MAX(ph.price) as highest_price_ever,
AVG(ph.price)::NUMERIC(12,2) as average_price,
PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY ph.price) as median_price,
PERCENTILE_CONT(0.25) WITHIN GROUP (ORDER BY ph.price) as q1_price,
PERCENTILE_CONT(0.75) WITHIN GROUP (ORDER BY ph.price) as q3_price,
STDDEV(ph.price)::NUMERIC(12,2) as price_stddev,
-- Current price (most recent year)
(SELECT price FROM price_history ph2
WHERE ph2.watch_id = w.id
ORDER BY year DESC LIMIT 1) as current_price,
-- Original price (earliest year)
(SELECT price FROM price_history ph2
WHERE ph2.watch_id = w.id
ORDER BY year ASC LIMIT 1) as original_price,
-- Appreciation calculations
CASE
WHEN MIN(ph.price) > 0 AND MAX(ph.price) > 0 THEN
((MAX(ph.price) - MIN(ph.price)) / MIN(ph.price) * 100)::NUMERIC(10,2)
ELSE NULL
END as appreciation_percentage,
-- Compound Annual Growth Rate (CAGR)
CASE
WHEN MIN(ph.price) > 0 AND (MAX(ph.year) - MIN(ph.year)) > 0 THEN
((POWER(MAX(ph.price) / MIN(ph.price), 1.0 / (MAX(ph.year) - MIN(ph.year))) - 1) * 100)::NUMERIC(10,2)
ELSE NULL
END as cagr_percentage,
-- Volatility metric
CASE
WHEN AVG(ph.price) > 0 THEN
(STDDEV(ph.price) / AVG(ph.price) * 100)::NUMERIC(10,2)
ELSE NULL
END as volatility_percentage,
-- Metadata
NOW() as calculated_at,
w.is_limited_edition,
w.is_featured
FROM watches w
LEFT JOIN price_history ph ON w.id = ph.watch_id
WHERE w.is_active = true
GROUP BY w.id, w.slug, w.model, w.series, w.reference, w.year_introduced,
w.is_limited_edition, w.is_featured;
-- Indexes on materialized view
CREATE UNIQUE INDEX idx_watch_statistics_id ON watch_statistics(id);
CREATE INDEX idx_watch_statistics_appreciation ON watch_statistics(appreciation_percentage DESC NULLS LAST);
CREATE INDEX idx_watch_statistics_cagr ON watch_statistics(cagr_percentage DESC NULLS LAST);
CREATE INDEX idx_watch_statistics_series ON watch_statistics(series);
CREATE INDEX idx_watch_statistics_current_price ON watch_statistics(current_price DESC NULLS LAST);
-- Series Performance: Aggregated series-level analytics
CREATE MATERIALIZED VIEW series_performance AS
SELECT
series,
COUNT(*) as watch_count,
AVG(appreciation_percentage)::NUMERIC(10,2) as avg_appreciation,
AVG(cagr_percentage)::NUMERIC(10,2) as avg_cagr,
AVG(current_price)::NUMERIC(12,2) as avg_current_price,
MIN(current_price) as min_current_price,
MAX(current_price) as max_current_price,
SUM(CASE WHEN appreciation_percentage > 0 THEN 1 ELSE 0 END) as appreciating_watches,
SUM(CASE WHEN is_limited_edition THEN 1 ELSE 0 END) as limited_edition_count,
NOW() as calculated_at
FROM watch_statistics
GROUP BY series;
CREATE UNIQUE INDEX idx_series_performance_series ON series_performance(series);
CREATE INDEX idx_series_performance_appreciation ON series_performance(avg_appreciation DESC);
-- Decade Performance: Historical price trends by decade
CREATE MATERIALIZED VIEW decade_performance AS
SELECT
(w.year_introduced / 10) * 10 as decade,
w.series,
COUNT(DISTINCT w.id) as watch_count,
AVG(ws.appreciation_percentage)::NUMERIC(10,2) as avg_appreciation,
AVG(ws.current_price)::NUMERIC(12,2) as avg_current_price,
MIN(w.year_introduced) as earliest_year,
MAX(w.year_introduced) as latest_year
FROM watches w
INNER JOIN watch_statistics ws ON w.id = ws.id
WHERE w.is_active = true
GROUP BY decade, w.series
ORDER BY decade DESC, w.series;
CREATE INDEX idx_decade_performance_decade ON decade_performance(decade);
CREATE INDEX idx_decade_performance_series ON decade_performance(series);
-- ============================================================================
-- USER ACTIVITY & ANALYTICS
-- ============================================================================
-- Watch Views: Tracking for trending analysis
CREATE TABLE watch_views (
id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
watch_id UUID NOT NULL REFERENCES watches(id) ON DELETE CASCADE,
session_id VARCHAR(255),
user_agent TEXT,
ip_address INET,
referrer TEXT,
country_code CHAR(2),
viewed_at TIMESTAMPTZ DEFAULT NOW(),
view_duration_seconds INTEGER
) PARTITION BY RANGE (viewed_at);
-- Create partitions by month for last 12 months + future
CREATE TABLE watch_views_2024_11 PARTITION OF watch_views
FOR VALUES FROM ('2024-11-01') TO ('2024-12-01');
CREATE TABLE watch_views_2024_12 PARTITION OF watch_views
FOR VALUES FROM ('2024-12-01') TO ('2025-01-01');
CREATE TABLE watch_views_2025_01 PARTITION OF watch_views
FOR VALUES FROM ('2025-01-01') TO ('2025-02-01');
CREATE TABLE watch_views_2025_02 PARTITION OF watch_views
FOR VALUES FROM ('2025-02-01') TO ('2025-03-01');
CREATE TABLE watch_views_2025_03 PARTITION OF watch_views
FOR VALUES FROM ('2025-03-01') TO ('2025-04-01');
CREATE INDEX idx_watch_views_watch_id ON watch_views(watch_id);
CREATE INDEX idx_watch_views_viewed_at ON watch_views(viewed_at DESC);
CREATE INDEX idx_watch_views_session ON watch_views(session_id);
-- User Watchlists: User-specific collections
CREATE TABLE user_watchlists (
id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
user_id VARCHAR(255) NOT NULL,
watch_id UUID NOT NULL REFERENCES watches(id) ON DELETE CASCADE,
notes TEXT,
target_price positive_price,
alert_enabled BOOLEAN DEFAULT false,
created_at TIMESTAMPTZ DEFAULT NOW(),
updated_at TIMESTAMPTZ DEFAULT NOW(),
UNIQUE(user_id, watch_id)
);
CREATE INDEX idx_watchlists_user ON user_watchlists(user_id);
CREATE INDEX idx_watchlists_watch ON user_watchlists(watch_id);
CREATE INDEX idx_watchlists_alerts ON user_watchlists(alert_enabled, target_price)
WHERE alert_enabled = true;
-- ============================================================================
-- DATA SOURCES & PROVENANCE
-- ============================================================================
CREATE TABLE data_sources (
id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
name VARCHAR(255) UNIQUE NOT NULL,
type VARCHAR(50) CHECK (type IN ('auction', 'dealer', 'catalog', 'estimate', 'community')),
url VARCHAR(500),
reliability_score INTEGER CHECK (reliability_score >= 1 AND reliability_score <= 5),
description TEXT,
is_active BOOLEAN DEFAULT true,
created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX idx_data_sources_type ON data_sources(type);
CREATE INDEX idx_data_sources_active ON data_sources(is_active) WHERE is_active = true;
-- Price History Sources: Many-to-many relationship
CREATE TABLE price_history_sources (
price_history_id UUID NOT NULL,
data_source_id UUID NOT NULL REFERENCES data_sources(id),
confidence_contribution NUMERIC(3,2) CHECK (confidence_contribution >= 0 AND confidence_contribution <= 1),
PRIMARY KEY (price_history_id, data_source_id)
);
-- ============================================================================
-- AUDIT & CHANGE TRACKING
-- ============================================================================
CREATE TABLE audit_log (
id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
table_name VARCHAR(100) NOT NULL,
record_id UUID NOT NULL,
operation VARCHAR(10) NOT NULL CHECK (operation IN ('INSERT', 'UPDATE', 'DELETE')),
old_values JSONB,
new_values JSONB,
changed_by VARCHAR(255),
changed_at TIMESTAMPTZ DEFAULT NOW(),
ip_address INET,
user_agent TEXT
) PARTITION BY RANGE (changed_at);
-- Partition by month
CREATE TABLE audit_log_2024_11 PARTITION OF audit_log
FOR VALUES FROM ('2024-11-01') TO ('2024-12-01');
CREATE TABLE audit_log_2024_12 PARTITION OF audit_log
FOR VALUES FROM ('2024-12-01') TO ('2025-01-01');
CREATE TABLE audit_log_2025_01 PARTITION OF audit_log
FOR VALUES FROM ('2025-01-01') TO ('2025-02-01');
CREATE INDEX idx_audit_log_table_record ON audit_log(table_name, record_id);
CREATE INDEX idx_audit_log_changed_at ON audit_log(changed_at DESC);
CREATE INDEX idx_audit_log_operation ON audit_log(operation);
-- ============================================================================
-- ETL PIPELINE SUPPORT
-- ============================================================================
-- ETL Job Tracking
CREATE TABLE etl_jobs (
id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
job_name VARCHAR(100) NOT NULL,
job_type VARCHAR(50) NOT NULL,
status VARCHAR(20) CHECK (status IN ('pending', 'running', 'completed', 'failed', 'cancelled')),
started_at TIMESTAMPTZ,
completed_at TIMESTAMPTZ,
records_processed INTEGER DEFAULT 0,
records_inserted INTEGER DEFAULT 0,
records_updated INTEGER DEFAULT 0,
records_failed INTEGER DEFAULT 0,
error_message TEXT,
execution_time_ms INTEGER,
metadata JSONB,
created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX idx_etl_jobs_status ON etl_jobs(status, started_at);
CREATE INDEX idx_etl_jobs_job_name ON etl_jobs(job_name);
CREATE INDEX idx_etl_jobs_created_at ON etl_jobs(created_at DESC);
-- Staging tables for ETL
CREATE TABLE staging_watches (
id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
source_data JSONB NOT NULL,
parsed_data JSONB,
validation_status VARCHAR(20),
validation_errors JSONB,
processed BOOLEAN DEFAULT false,
created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX idx_staging_watches_processed ON staging_watches(processed) WHERE processed = false;
-- ============================================================================
-- TRIGGERS & AUTOMATION
-- ============================================================================
-- Auto-update timestamps
CREATE OR REPLACE FUNCTION update_timestamp()
RETURNS TRIGGER AS $$
BEGIN
NEW.updated_at = NOW();
RETURN NEW;
END;
$$ LANGUAGE plpgsql;
CREATE TRIGGER watches_update_timestamp
BEFORE UPDATE ON watches
FOR EACH ROW EXECUTE FUNCTION update_timestamp();
CREATE TRIGGER specifications_update_timestamp
BEFORE UPDATE ON specifications
FOR EACH ROW EXECUTE FUNCTION update_timestamp();
CREATE TRIGGER user_watchlists_update_timestamp
BEFORE UPDATE ON user_watchlists
FOR EACH ROW EXECUTE FUNCTION update_timestamp();
-- Full-text search vector update
CREATE OR REPLACE FUNCTION update_watch_search_vector()
RETURNS TRIGGER AS $$
BEGIN
NEW.search_vector :=
setweight(to_tsvector('english', COALESCE(NEW.model, '')), 'A') ||
setweight(to_tsvector('english', COALESCE(NEW.series::text, '')), 'A') ||
setweight(to_tsvector('english', COALESCE(NEW.reference, '')), 'B') ||
setweight(to_tsvector('english', COALESCE(NEW.description, '')), 'C') ||
setweight(to_tsvector('english', COALESCE(NEW.detailed_description, '')), 'D');
RETURN NEW;
END;
$$ LANGUAGE plpgsql;
CREATE TRIGGER watches_search_vector_update
BEFORE INSERT OR UPDATE ON watches
FOR EACH ROW EXECUTE FUNCTION update_watch_search_vector();
-- Audit trail trigger
CREATE OR REPLACE FUNCTION audit_trigger_function()
RETURNS TRIGGER AS $$
BEGIN
IF TG_OP = 'DELETE' THEN
INSERT INTO audit_log (table_name, record_id, operation, old_values, changed_by)
VALUES (TG_TABLE_NAME, OLD.id, TG_OP, row_to_json(OLD), current_user);
RETURN OLD;
ELSIF TG_OP = 'UPDATE' THEN
INSERT INTO audit_log (table_name, record_id, operation, old_values, new_values, changed_by)
VALUES (TG_TABLE_NAME, NEW.id, TG_OP, row_to_json(OLD), row_to_json(NEW), current_user);
RETURN NEW;
ELSIF TG_OP = 'INSERT' THEN
INSERT INTO audit_log (table_name, record_id, operation, new_values, changed_by)
VALUES (TG_TABLE_NAME, NEW.id, TG_OP, row_to_json(NEW), current_user);
RETURN NEW;
END IF;
END;
$$ LANGUAGE plpgsql;
CREATE TRIGGER watches_audit_trigger
AFTER INSERT OR UPDATE OR DELETE ON watches
FOR EACH ROW EXECUTE FUNCTION audit_trigger_function();
CREATE TRIGGER price_history_audit_trigger
AFTER INSERT OR UPDATE OR DELETE ON price_history
FOR EACH ROW EXECUTE FUNCTION audit_trigger_function();
-- ============================================================================
-- FUNCTIONS & STORED PROCEDURES
-- ============================================================================
-- Refresh all materialized views
CREATE OR REPLACE FUNCTION refresh_all_materialized_views()
RETURNS void AS $$
BEGIN
REFRESH MATERIALIZED VIEW CONCURRENTLY watch_statistics;
REFRESH MATERIALIZED VIEW CONCURRENTLY series_performance;
REFRESH MATERIALIZED VIEW CONCURRENTLY decade_performance;
END;
$$ LANGUAGE plpgsql;
-- Get trending watches
CREATE OR REPLACE FUNCTION get_trending_watches(
days_back INTEGER DEFAULT 7,
limit_count INTEGER DEFAULT 10
)
RETURNS TABLE (
watch_id UUID,
slug VARCHAR(255),
model VARCHAR(255),
series watch_series_enum,
view_count BIGINT,
avg_duration NUMERIC(10,2)
) AS $$
BEGIN
RETURN QUERY
SELECT
w.id,
w.slug,
w.model,
w.series,
COUNT(*) as view_count,
AVG(wv.view_duration_seconds)::NUMERIC(10,2) as avg_duration
FROM watches w
INNER JOIN watch_views wv ON w.id = wv.watch_id
WHERE wv.viewed_at >= NOW() - (days_back || ' days')::INTERVAL
AND w.is_active = true
GROUP BY w.id, w.slug, w.model, w.series
ORDER BY view_count DESC
LIMIT limit_count;
END;
$$ LANGUAGE plpgsql;
-- Full-text search with ranking
CREATE OR REPLACE FUNCTION search_watches(
search_query TEXT,
limit_count INTEGER DEFAULT 20
)
RETURNS TABLE (
watch_id UUID,
slug VARCHAR(255),
model VARCHAR(255),
series watch_series_enum,
rank REAL
) AS $$
BEGIN
RETURN QUERY
SELECT
w.id,
w.slug,
w.model,
w.series,
ts_rank(w.search_vector, query) as rank
FROM watches w,
to_tsquery('english', search_query) query
WHERE w.search_vector @@ query
AND w.is_active = true
ORDER BY rank DESC
LIMIT limit_count;
END;
$$ LANGUAGE plpgsql;
-- Calculate price appreciation between years
CREATE OR REPLACE FUNCTION calculate_appreciation(
p_watch_id UUID,
p_start_year INTEGER,
p_end_year INTEGER
)
RETURNS TABLE (
start_price NUMERIC,
end_price NUMERIC,
appreciation_pct NUMERIC,
cagr NUMERIC
) AS $$
BEGIN
RETURN QUERY
WITH price_points AS (
SELECT
(SELECT price FROM price_history
WHERE watch_id = p_watch_id AND year = p_start_year
ORDER BY year LIMIT 1) as start_price,
(SELECT price FROM price_history
WHERE watch_id = p_watch_id AND year = p_end_year
ORDER BY year DESC LIMIT 1) as end_price
)
SELECT
pp.start_price,
pp.end_price,
CASE
WHEN pp.start_price > 0 THEN
((pp.end_price - pp.start_price) / pp.start_price * 100)::NUMERIC(10,2)
ELSE NULL
END as appreciation_pct,
CASE
WHEN pp.start_price > 0 AND (p_end_year - p_start_year) > 0 THEN
((POWER(pp.end_price / pp.start_price, 1.0 / (p_end_year - p_start_year)) - 1) * 100)::NUMERIC(10,2)
ELSE NULL
END as cagr
FROM price_points pp;
END;
$$ LANGUAGE plpgsql;
-- Get watch details with all related data
CREATE OR REPLACE FUNCTION get_watch_details(p_watch_id UUID)
RETURNS JSON AS $$
DECLARE
result JSON;
BEGIN
SELECT json_build_object(
'watch', row_to_json(w),
'specifications', row_to_json(s),
'features', (
SELECT json_agg(f.name ORDER BY wf.display_order)
FROM watch_features wf
INNER JOIN features f ON wf.feature_id = f.id
WHERE wf.watch_id = w.id
),
'price_history', (
SELECT json_agg(row_to_json(ph) ORDER BY ph.year)
FROM price_history ph
WHERE ph.watch_id = w.id
),
'statistics', row_to_json(ws)
) INTO result
FROM watches w
LEFT JOIN specifications s ON w.id = s.watch_id
LEFT JOIN watch_statistics ws ON w.id = ws.id
WHERE w.id = p_watch_id;
RETURN result;
END;
$$ LANGUAGE plpgsql;
-- ============================================================================
-- INITIAL DATA SOURCES
-- ============================================================================
INSERT INTO data_sources (name, type, reliability_score, description) VALUES
('Christie''s Speedmaster 50 Auction', 'auction', 5, 'Major auction house - highly reliable'),
('Sotheby''s Omega Collection', 'auction', 5, 'Major auction house - highly reliable'),
('Phillips Watches Auction', 'auction', 5, 'Specialist watch auction house'),
('Chrono24 Market Data', 'dealer', 4, 'Leading watch marketplace'),
('WatchCharts Index', 'estimate', 4, 'Market index and analytics'),
('Historical Catalogs', 'catalog', 5, 'Official manufacturer catalogs'),
('Omega Museum Archives', 'catalog', 5, 'Official historical records');
-- ============================================================================
-- PERFORMANCE OPTIMIZATION
-- ============================================================================
-- Analyze all tables for query planner
ANALYZE watches;
ANALYZE specifications;
ANALYZE features;
ANALYZE watch_features;
ANALYZE price_history;
ANALYZE watch_views;
ANALYZE user_watchlists;
ANALYZE data_sources;
-- Vacuum for optimal performance
VACUUM ANALYZE;
-- ============================================================================
-- COMMENTS & DOCUMENTATION
-- ============================================================================
COMMENT ON SCHEMA omega IS 'Omega Watch Price History Tracking System - Enterprise Edition';
COMMENT ON TABLE watches IS 'Main watch catalog with comprehensive metadata';
COMMENT ON TABLE price_history IS 'Historical price tracking (partitioned by year for scalability)';
COMMENT ON TABLE specifications IS 'Technical specifications with extracted numeric values';
COMMENT ON MATERIALIZED VIEW watch_statistics IS 'Pre-calculated statistics for high-performance queries';
COMMENT ON MATERIALIZED VIEW series_performance IS 'Series-level aggregated analytics';
COMMENT ON FUNCTION refresh_all_materialized_views() IS 'Refresh all materialized views concurrently';
COMMENT ON FUNCTION get_trending_watches(INTEGER, INTEGER) IS 'Get most viewed watches in recent days';
COMMENT ON FUNCTION search_watches(TEXT, INTEGER) IS 'Full-text search with ranking';
-- Reset search path
RESET search_path;