tsapi-seriesdb-shared-sql/sql/05_procedures.sql

-- 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$;