Rules 與 LISTEN/NOTIFY：查詢改寫與即時事件通知 | PostgreSQL

CREATE [ OR REPLACE ] RULE rule_name AS
    ON { SELECT | INSERT | UPDATE | DELETE }
    TO table_name
    [ WHERE condition ]
    DO [ ALSO | INSTEAD ] { NOTHING | command | ( command ; command ... ) }

-- 建立歸檔資料表
CREATE TABLE orders_archive (LIKE orders INCLUDING ALL);

-- 所有 DELETE 改為移動到歸檔表
CREATE OR REPLACE RULE archive_on_delete AS
    ON DELETE TO orders
    DO INSTEAD (
        INSERT INTO orders_archive
            SELECT OLD.*;
    );

-- 執行 DELETE 時，資料不會被刪除，而是複製到 orders_archive
DELETE FROM orders WHERE id = 42;

CREATE TABLE orders_audit_log (
    audit_id   BIGSERIAL PRIMARY KEY,
    order_id   INT,
    action     TEXT,
    logged_at  TIMESTAMPTZ DEFAULT now()
);

CREATE OR REPLACE RULE log_on_insert AS
    ON INSERT TO orders
    DO ALSO (
        INSERT INTO orders_audit_log (order_id, action)
        VALUES (NEW.id, 'INSERT');
    );

-- 禁止刪除狀態為 active 的訂單
CREATE OR REPLACE RULE protect_active_orders AS
    ON DELETE TO orders
    WHERE (OLD.status = 'active')
    DO INSTEAD NOTHING;

-- 建立底層資料表
CREATE TABLE employees (
    id         SERIAL PRIMARY KEY,
    name       TEXT NOT NULL,
    department TEXT NOT NULL,
    salary     NUMERIC,
    is_active  BOOLEAN DEFAULT true
);

-- 只顯示在職員工的 View
CREATE VIEW active_employees AS
    SELECT id, name, department, salary
    FROM employees
    WHERE is_active = true;

-- 讓 View 支援 INSERT（自動設定 is_active = true）
CREATE OR REPLACE RULE active_employees_insert AS
    ON INSERT TO active_employees
    DO INSTEAD (
        INSERT INTO employees (name, department, salary, is_active)
        VALUES (NEW.name, NEW.department, NEW.salary, true);
    );

-- 讓 View 的 DELETE 轉為軟刪除
CREATE OR REPLACE RULE active_employees_delete AS
    ON DELETE TO active_employees
    DO INSTEAD (
        UPDATE employees
        SET is_active = false
        WHERE id = OLD.id;
    );

-- 訂閱者（連線A）
LISTEN order_events;

-- 發布者（連線B）
NOTIFY order_events;

-- 發送含 Payload 的通知（通常使用 JSON 格式）
NOTIFY order_events, '{"action":"created","order_id":42}';

BEGIN;

INSERT INTO orders (customer_id, total_amount, status)
VALUES (101, 1500.00, 'pending')
RETURNING id;

-- NOTIFY 被記錄，但要等到 COMMIT 後才真正送出
NOTIFY order_events, '{"action":"order_created","order_id":123}';

COMMIT;
-- 此時訂閱者才會收到通知

-- 通用事件通知 Trigger 函式
CREATE OR REPLACE FUNCTION notify_table_change()
RETURNS TRIGGER AS $$
DECLARE
    payload JSON;
    record_id BIGINT;
BEGIN
    IF TG_OP = 'DELETE' THEN
        record_id := OLD.id;
    ELSE
        record_id := NEW.id;
    END IF;

    -- 組裝 JSON Payload
    payload := json_build_object(
        'table',     TG_TABLE_NAME,
        'schema',    TG_TABLE_SCHEMA,
        'action',    TG_OP,
        'id',        record_id,
        'timestamp', EXTRACT(EPOCH FROM clock_timestamp())
    );

    -- 發送通知到以資料表名稱命名的頻道
    PERFORM pg_notify(TG_TABLE_NAME || '_events', payload::TEXT);

    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

-- 將 Trigger 綁定到 orders 資料表
CREATE TRIGGER orders_notify_trigger
    AFTER INSERT OR UPDATE OR DELETE ON orders
    FOR EACH ROW
    EXECUTE FUNCTION notify_table_change();

-- 推薦的 Payload 結構
CREATE OR REPLACE FUNCTION build_event_payload(
    p_action     TEXT,
    p_entity     TEXT,
    p_entity_id  BIGINT,
    p_data       JSONB DEFAULT NULL
) RETURNS TEXT AS $$
BEGIN
    RETURN json_build_object(
        'version',   1,                                    -- 版本號
        'action',    p_action,                             -- created / updated / deleted
        'entity',    p_entity,                             -- order / user / product
        'entity_id', p_entity_id,
        'data',      COALESCE(p_data, 'null'::JSONB),
        'timestamp', to_char(now() AT TIME ZONE 'UTC',
                             'YYYY-MM-DD"T"HH24:MI:SS"Z"')
    )::TEXT;
END;
$$ LANGUAGE plpgsql;

const { Client } = require('pg');

async function startOrderListener() {
    // LISTEN/NOTIFY 必須使用獨立的長連線，不能使用 Pool
    const client = new Client({
        connectionString: process.env.DATABASE_URL,
    });

    await client.connect();
    await client.query('LISTEN order_events');

    // 綁定通知事件處理器
    client.on('notification', async (msg) => {
        try {
            const event = JSON.parse(msg.payload);
            console.log(`收到 ${msg.channel} 事件:`, event);

            switch (event.action) {
                case 'INSERT':
                    await handleOrderCreated(event);
                    break;
                case 'UPDATE':
                    await handleOrderUpdated(event);
                    break;
                case 'DELETE':
                    await handleOrderDeleted(event);
                    break;
            }
        } catch (err) {
            console.error('處理通知時發生錯誤:', err);
        }
    });

    // 處理連線錯誤與重連
    client.on('error', async (err) => {
        console.error('資料庫連線錯誤:', err);
        setTimeout(() => startOrderListener(), 5000);
    });
}

startOrderListener().catch(console.error);

import psycopg2
import psycopg2.extensions
import select
import json

def listen_for_events(dsn: str, channel: str):
    conn = psycopg2.connect(dsn)
    # 必須設定 AUTOCOMMIT
    conn.set_isolation_level(
        psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT
    )

    cursor = conn.cursor()
    cursor.execute(f"LISTEN {channel};")
    print(f"開始監聽頻道：{channel}")

    try:
        while True:
            if select.select([conn], [], [], 5) == ([], [], []):
                print("等待中... (心跳)")
            else:
                conn.poll()
                while conn.notifies:
                    notify = conn.notifies.pop(0)
                    payload = json.loads(notify.payload)
                    print(f"收到通知 | PID: {notify.pid}")
                    process_event(payload)
    finally:
        conn.close()

import asyncio
import json
import psycopg

async def listen_async(conninfo: str, channel: str):
    async with await psycopg.AsyncConnection.connect(
        conninfo, autocommit=True
    ) as conn:
        await conn.execute(f"LISTEN {channel}")
        async for notify in conn.notifies():
            payload = json.loads(notify.payload)
            print(f"收到通知：{payload}")
            await process_event_async(payload)

-- 資料更新後通知快取層失效
CREATE OR REPLACE FUNCTION invalidate_cache_on_change()
RETURNS TRIGGER AS $$
BEGIN
    PERFORM pg_notify(
        'cache_invalidation',
        json_build_object(
            'entity', TG_TABLE_NAME,
            'id',     COALESCE(NEW.id, OLD.id),
            'keys',   ARRAY[
                TG_TABLE_NAME || ':' || COALESCE(NEW.id, OLD.id)::TEXT,
                TG_TABLE_NAME || ':list'
            ]
        )::TEXT
    );
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;