-- Stored procedures and functions -- Generated: 2025-10-15T12:24:56.582392 -- Log schema functions -- Function: log.notify_event_log -- Writes a breadcrumb to the staging table (UNLOGGED, fast). -- Self-drains every 20,000th insert to prevent unbounded growth. CREATE OR REPLACE FUNCTION log.notify_event_log( p_series_name TEXT, p_db_schema TEXT, p_op_type TEXT, p_details JSONB DEFAULT NULL ) RETURNS void LANGUAGE plpgsql AS $function$ BEGIN INSERT INTO log.pending_changes (series_name, db_schema, op_type, details) VALUES (p_series_name, p_db_schema, p_op_type, p_details); -- Self-drain every 20,000th insert IF (currval('log.pending_changes_id_seq') % 20000 = 0) THEN PERFORM log.flush_pending_to_event_log(); END IF; EXCEPTION WHEN OTHERS THEN RAISE WARNING 'notify_event_log failed: %', SQLERRM; END; $function$; -- Function: log.flush_pending_to_event_log -- Drains the staging table into the permanent event_log. -- Safe for concurrent use: snapshots max(id) before processing. -- After moving events, verifies that meta.series timestamps match actual -- data for any series that had data-modifying events. If a mismatch is -- found, writes a TIMESTAMP_MISMATCH event to log.event_log for visibility. -- (Timestamps are kept accurate by the application on every data operation, -- so this step is a safety net, not a correction mechanism.) CREATE OR REPLACE FUNCTION log.flush_pending_to_event_log() RETURNS INT LANGUAGE plpgsql AS $function$ DECLARE batch_id BIGINT; v_count INT; _rec RECORD; _actual_first TIMESTAMP; _actual_last TIMESTAMP; _meta_first TIMESTAMP; _meta_last TIMESTAMP; BEGIN -- Serialise flushes. Snapshotting MAX(id) does not make this safe on its -- own: two callers can both read the same batch_id, both copy those rows -- to log.event_log, and only then delete them -- duplicating permanent -- audit events. The lock is transaction-scoped, so it is released when -- this function's transaction ends, however it ends. PERFORM pg_advisory_xact_lock(hashtext('tsapi_pgdb.flush_event_log')); SELECT COALESCE(MAX(id), 0) INTO batch_id FROM log.pending_changes; IF batch_id = 0 THEN RETURN 0; END IF; -- 1. Move rows from staging to permanent event_log INSERT INTO log.event_log ("timestamp", series_name, db_schema, event_type, details, source, db_user) SELECT "timestamp", series_name, db_schema, op_type, details, 'app', db_user FROM log.pending_changes WHERE id <= batch_id ORDER BY id; GET DIAGNOSTICS v_count = ROW_COUNT; -- 2. Verify meta.series timestamps for series with data-modifying events FOR _rec IN SELECT DISTINCT series_name, db_schema FROM log.pending_changes WHERE id <= batch_id AND op_type IN ( 'CSV_IMPORT', 'JSON_APPEND', 'DATAFRAME_IMPORT', 'DELETE_FROM', 'DELETE_AFTER', 'DELETE_RANGE', 'DELETE_UNTIL', 'DELETE_WEEKENDS', 'DELETE_TIME_RANGE', 'SERIES_TRUNCATED' ) LOOP BEGIN EXECUTE format( 'SELECT MIN(timestamp), MAX(timestamp) FROM %I.%I', _rec.db_schema, _rec.series_name ) INTO _actual_first, _actual_last; SELECT first_timestamp, last_timestamp INTO _meta_first, _meta_last FROM meta.series WHERE table_name = _rec.series_name AND db_schema = _rec.db_schema; IF _meta_first IS DISTINCT FROM _actual_first OR _meta_last IS DISTINCT FROM _actual_last THEN INSERT INTO log.event_log ("timestamp", series_name, db_schema, event_type, details, source, db_user) VALUES ( CURRENT_TIMESTAMP, _rec.series_name, _rec.db_schema, 'TIMESTAMP_MISMATCH', jsonb_build_object( 'meta_first', _meta_first, 'meta_last', _meta_last, 'actual_first', _actual_first, 'actual_last', _actual_last ), 'flush_verify', CURRENT_USER ); END IF; EXCEPTION WHEN OTHERS THEN NULL; -- table may have been dropped in same batch END; END LOOP; -- 3. Clear staging DELETE FROM log.pending_changes WHERE id <= batch_id; RETURN v_count; END; $function$; -- Meta schema functions -- Function: meta.update_series_last_timestamp -- Lightweight timestamp update for append operations. -- Uses GREATEST to conditionally update last_timestamp (no table scan). -- Sets first_timestamp on first insert via COALESCE. -- p_first_timestamp is optional: when NULL, uses p_last_timestamp for first_timestamp. -- For single inserts, only p_last_timestamp is needed. -- For batch inserts, pass min timestamp as p_first_timestamp. -- Both Python and C++ clients call this after appending data. CREATE OR REPLACE FUNCTION meta.update_series_last_timestamp( p_table_name TEXT, p_db_schema TEXT DEFAULT 'series', p_last_timestamp TIMESTAMP DEFAULT NULL, p_first_timestamp TIMESTAMP DEFAULT NULL ) RETURNS void LANGUAGE plpgsql AS $function$ BEGIN UPDATE meta.series SET first_timestamp = COALESCE(first_timestamp, p_first_timestamp, p_last_timestamp), last_timestamp = GREATEST(last_timestamp, p_last_timestamp) WHERE table_name = p_table_name AND db_schema = p_db_schema; END; $function$; -- Function: meta.refresh_series_timestamps -- Full timestamp refresh using MIN/MAX scan. -- Used after deletes, truncates, or any operation that may change first_timestamp. -- BRIN indexes make MIN/MAX fast even on large tables. -- Both Python and C++ clients call this after data deletion. CREATE OR REPLACE FUNCTION meta.refresh_series_timestamps( p_table_name TEXT, p_db_schema TEXT DEFAULT 'series' ) RETURNS void LANGUAGE plpgsql AS $function$ DECLARE _first_ts TIMESTAMP; _last_ts TIMESTAMP; BEGIN EXECUTE format( 'SELECT MIN(timestamp), MAX(timestamp) FROM %I.%I', p_db_schema, p_table_name ) INTO _first_ts, _last_ts; UPDATE meta.series SET first_timestamp = _first_ts, last_timestamp = _last_ts WHERE table_name = p_table_name AND db_schema = p_db_schema; END; $function$; -- Function: meta.drop_all_series CREATE OR REPLACE FUNCTION meta.drop_all_series() RETURNS void LANGUAGE plpgsql AS $function$ DECLARE _rec record; BEGIN -- Set the context to allow drops PERFORM set_config('application.dropping_series', 'true', true); -- Loop through all series in series table FOR _rec IN SELECT "table_name", "db_schema" FROM meta.series LOOP -- Drop each table EXECUTE format('DROP TABLE IF EXISTS %I.%I', _rec."db_schema", _rec."table_name"); END LOOP; -- Clear the series table DELETE FROM meta.series; -- Reset the context PERFORM set_config('application.dropping_series', 'false', true); END; $function$; -- Function: meta.drop_series CREATE OR REPLACE FUNCTION meta.drop_series(_table_name text, _schema_name text DEFAULT 'series'::text, _raise_if_missing boolean DEFAULT true) RETURNS void LANGUAGE plpgsql AS $function$ BEGIN -- Check if the series exists in series table IF NOT EXISTS ( SELECT 1 FROM meta.series WHERE "table_name" = _table_name AND "db_schema" = _schema_name ) THEN IF _raise_if_missing THEN RAISE EXCEPTION 'Series %.% does not exist', _schema_name, _table_name; END IF; RETURN; END IF; -- Set the context to allow drops PERFORM set_config('application.dropping_series', 'true', true); -- Drop the table EXECUTE format('DROP TABLE IF EXISTS %I.%I', _schema_name, _table_name); -- Remove from series table DELETE FROM meta.series WHERE "table_name" = _table_name AND "db_schema" = _schema_name; -- Reset the context PERFORM set_config('application.dropping_series', 'false', true); END; $function$; -- Function: meta.drop_series_if_exists CREATE OR REPLACE FUNCTION meta.drop_series_if_exists(_table_name text, _schema_name text DEFAULT 'series'::text) RETURNS void LANGUAGE plpgsql AS $function$ BEGIN -- Check if the series exists in series table IF EXISTS ( SELECT 1 FROM meta.series WHERE "table_name" = _table_name AND "db_schema" = _schema_name ) THEN -- Signal the prevent_direct_drop event trigger that this DROP is -- coming from a sanctioned procedure. Without it the trigger rejects -- the statement below, because the table is registered in meta.series. -- Local to the transaction, so it cannot leak to later statements. PERFORM set_config('application.dropping_series', 'true', true); -- Drop the table EXECUTE format('DROP TABLE IF EXISTS %I.%I', _schema_name, _table_name); -- Remove from series table DELETE FROM meta.series WHERE "table_name" = _table_name AND "db_schema" = _schema_name; PERFORM set_config('application.dropping_series', 'false', true); END IF; END; $function$; -- Function: meta.prevent_direct_drop CREATE OR REPLACE FUNCTION meta.prevent_direct_drop() RETURNS event_trigger LANGUAGE plpgsql AS $function$ DECLARE obj record; BEGIN -- Get the current context IF current_setting('application.dropping_series', true) = 'true' THEN -- Allow the drop if it's coming from our procedures RETURN; END IF; FOR obj IN SELECT * FROM pg_event_trigger_dropped_objects() WHERE object_type = 'table' LOOP IF EXISTS ( SELECT 1 FROM meta.series WHERE table_name = obj.object_name AND db_schema = obj.schema_name ) THEN RAISE EXCEPTION 'Table %.% is registered in series and can only be dropped using TimeseriesDB.drop_series()', obj.schema_name, obj.object_name; END IF; END LOOP; END; $function$; -- Function: meta.read_series_batch CREATE OR REPLACE FUNCTION meta.read_series_batch(p_schema_name text, p_table_name text, p_start_ts timestamp without time zone DEFAULT NULL::timestamp without time zone, p_batch_size integer DEFAULT 10000) RETURNS TABLE(result_json jsonb) LANGUAGE plpgsql AS $function$ DECLARE v_query TEXT; v_qualified_name TEXT; BEGIN -- Build qualified table name (with proper quoting) v_qualified_name := format('%I.%I', p_schema_name, p_table_name); -- Build query with proper WHERE clause IF p_start_ts IS NULL THEN -- Read from beginning v_query := format(' SELECT to_jsonb(t.*) AS result_json FROM ( SELECT * FROM %s ORDER BY timestamp LIMIT %s ) t ', v_qualified_name, p_batch_size); ELSE -- Read from specified timestamp (inclusive) v_query := format(' SELECT to_jsonb(t.*) AS result_json FROM ( SELECT * FROM %s WHERE timestamp >= %L ORDER BY timestamp LIMIT %s ) t ', v_qualified_name, p_start_ts, p_batch_size); END IF; -- Execute and return RETURN QUERY EXECUTE v_query; END; $function$; -- Function: meta.read_series_batch_cursor CREATE OR REPLACE FUNCTION meta.read_series_batch_cursor(p_schema_name text, p_table_name text, p_start_ts timestamp without time zone DEFAULT NULL::timestamp without time zone, p_batch_size integer DEFAULT 10000, p_cursor_name text DEFAULT 'series_cursor'::text) RETURNS refcursor LANGUAGE plpgsql AS $function$ DECLARE v_query TEXT; v_qualified_name TEXT; v_cursor refcursor; BEGIN -- Build qualified table name (with proper quoting) v_qualified_name := format('%I.%I', p_schema_name, p_table_name); -- Open cursor with name v_cursor := p_cursor_name; -- Build query with proper WHERE clause IF p_start_ts IS NULL THEN -- Read from beginning v_query := format(' SELECT * FROM %s ORDER BY timestamp LIMIT %s ', v_qualified_name, p_batch_size); ELSE -- Read from specified timestamp (inclusive) v_query := format(' SELECT * FROM %s WHERE timestamp >= %L ORDER BY timestamp LIMIT %s ', v_qualified_name, p_start_ts, p_batch_size); END IF; -- Open cursor OPEN v_cursor FOR EXECUTE v_query; RETURN v_cursor; END; $function$; -- Function: meta.read_series_batch_ohlc CREATE OR REPLACE FUNCTION meta.read_series_batch_ohlc(p_schema_name text, p_table_name text, p_start_ts timestamp without time zone DEFAULT NULL::timestamp without time zone, p_batch_size integer DEFAULT 10000) RETURNS TABLE(ts timestamp without time zone, open double precision, high double precision, low double precision, close double precision, volume bigint) LANGUAGE plpgsql AS $function$ DECLARE v_query TEXT; v_qualified_name TEXT; BEGIN -- Build qualified table name (with proper quoting) v_qualified_name := format('%I.%I', p_schema_name, p_table_name); -- Build query with proper WHERE clause IF p_start_ts IS NULL THEN -- Read from beginning v_query := format(' SELECT timestamp AS ts, open, high, low, close, volume FROM %s ORDER BY timestamp LIMIT %s ', v_qualified_name, p_batch_size); ELSE -- Read from specified timestamp (inclusive) v_query := format(' SELECT timestamp AS ts, open, high, low, close, volume FROM %s WHERE timestamp >= %L ORDER BY timestamp LIMIT %s ', v_qualified_name, p_start_ts, p_batch_size); END IF; -- Execute and return RETURN QUERY EXECUTE v_query; END; $function$;