unzip litmus_pgnats-*.zip

sudo ./install.sh

sudo ./install.sh /usr/lib/postgresql/18/bin/pg_config

CREATE EXTENSION IF NOT EXISTS litmus_pgnats;

CREATE SERVER le_nats_fdw_server FOREIGN DATA WRAPPER litmus_pgnats_fdw OPTIONS (
    host '<Litmus Edge IP>',
    port '4222',
    capacity '128',
    username '',
    password '<Access Account token>'
);

CREATE SERVER nats_fdw_server FOREIGN DATA WRAPPER litmus_pgnats_fdw OPTIONS (
    -- IP/hostname of the NATS message server (default: 127.0.0.1)
    host 'localhost',

    -- TCP port for NATS connections (default: 4222)
    port '4222',

    -- Internal command buffer size in messages (default: 128)
    capacity '128',

     -- Path to the CA certificate used to verify the NATS server certificate.
    -- Required for TLS when tls_insecure is not set.
    tls_ca_path '/path/ca',

    -- Path to the client certificate for mutual TLS authentication.
    -- Optional unless the server requires client auth.
    tls_cert_path '/path/cert',

    -- Path to the client private key. Required when tls_cert_path is set.
    tls_key_path '/path/key',

    --optional:
    -- Skip TLS certificate verification entirely.
    -- For development or servers with self-signed/private-CA certs (e.g. Litmus Edge).
    tls_insecure 'true',

    --optional:
    -- Default JetStream domain for all KV, object store, and stream publish calls.
    -- Can be overridden per-call via the domain parameter on individual functions.
    domain 'hub',

    --optional:
    -- NATS username for username/password authentication.
    -- Requires password or password_path.
    username 'myuser',

    --optional:
    -- Absolute path to a file containing the NATS password.
    -- File contents are trimmed of surrounding whitespace.
    -- Preferred over the inline password option — rotation takes effect on reconnect.
    password_path '/run/secrets/nats_password',

    --optional:
    -- Inline NATS password. Stored in pg_foreign_server.options (visible to superusers).
    password 'hunter2',

    --optional:
    -- Absolute path to a file containing an NKey seed string.
    -- Takes precedence over username/password when set.
    nkey_seed_path '/run/secrets/nats_nkey_seed',

    --optional:
    -- Inline NKey seed. Stored in pg_foreign_server.options (visible to superusers).
    nkey_seed 'SUABC...',

    --optional:
    -- Absolute path to a file containing a NATS token (no username required).
    -- Takes precedence over username/password; use for NATS token auth mode.
    token_path '/run/secrets/nats_token',

    --optional:
    -- Inline NATS token. Stored in pg_foreign_server.options (visible to superusers).
    token 'my-token',

    --optional:
    -- NATS subject for Patroni role-change notifications.
    notify_subject 'my.subject',

    --optional:
    -- Patroni REST API URL for role-change detection.
    patroni_url 'http://localhost:8008/patroni',

    --optional:
    -- Statement timeout (ms) applied to responder handler SPI calls (default: 5000).
    responder_timeout_ms '5000'
);

-- Core NATS (fire and forget)
SELECT nats_publish_text('sub.ject', 'hello');
SELECT nats_publish_json('sub.ject', '{"key": "value"}'::json);
SELECT nats_publish_jsonb('sub.ject', '{"key": "value"}'::jsonb);
SELECT nats_publish_binary('sub.ject', 'data'::bytea);

-- With reply subject
SELECT nats_publish_text('sub.ject', 'hello', 'reply.subject');

-- With headers
SELECT nats_publish_text('sub.ject', 'hello', NULL, '{"Nats-Msg-Id": "1"}'::jsonb);

-- JetStream (sync)
SELECT nats_publish_text_stream('sub.ject', 'hello');
SELECT nats_publish_json_stream('sub.ject', '{"key": "value"}'::json);
SELECT nats_publish_jsonb_stream('sub.ject', '{"key": "value"}'::jsonb);
SELECT nats_publish_binary_stream('sub.ject', 'data'::bytea);

-- JetStream with domain override
SELECT nats_publish_text_stream('sub.ject', 'hello', NULL, 'hub');

-- Subscribe a PostgreSQL function to a NATS subject
SELECT nats_subscribe('events.user.created', 'schema.handle_user_created'::regproc);

-- Multiple functions can be subscribed to the same subject
SELECT nats_subscribe('events.user.created', 'schema.log_user_created'::regproc);

-- Unsubscribe
SELECT nats_unsubscribe('events.user.created', 'schema.handle_user_created'::regproc);

SELECT nats_request_text('rpc.subject', 'payload', 5000);    -- timeout in ms
SELECT nats_request_json('rpc.subject', '{"q": "v"}'::json, 5000);
SELECT nats_request_jsonb('rpc.subject', '{"q": "v"}'::jsonb, 5000);
SELECT nats_request_binary('rpc.subject', 'data'::bytea, 5000);

SELECT litmus_pgnats_register_responder('rpc.ping', 'public.handle_ping'::regproc);
SELECT litmus_pgnats_unregister_responder('rpc.ping');

-- simple responder function
CREATE OR REPLACE FUNCTION litmus_pgnats.test_responder(
	  payload  text,
	  subject  text,
	  reply_to text
)
RETURNS text LANGUAGE sql AS $$
	  SELECT 'echo: ' || payload;
$$;
	
-- adding responder
SELECT litmus_pgnats_register_responder('litmus.test.responder', 'litmus_pgnats.test_responder'::regproc);

-- Write
SELECT nats_put_text('bucket', 'key', 'value');
SELECT nats_put_json('bucket', 'key', '{"x": 1}'::json);
SELECT nats_put_jsonb('bucket', 'key', '{"x": 1}'::jsonb);
SELECT nats_put_binary('bucket', 'key', 'data'::bytea);

-- Write to a specific domain
SELECT nats_put_text('bucket', 'key', 'value', 'hub');

-- Read
SELECT nats_get_text('bucket', 'key');
SELECT nats_get_json('bucket', 'key');
SELECT nats_get_jsonb('bucket', 'key');
SELECT nats_get_binary('bucket', 'key');

-- Read from a specific domain
SELECT nats_get_text('bucket', 'key', 'hub');

-- Delete
SELECT nats_delete_value('bucket', 'key');
SELECT nats_delete_value('bucket', 'key', 'hub');

-- Upload
SELECT nats_put_file('store', 'file.txt', 'content'::bytea);
SELECT nats_put_file('store', 'file.txt', 'content'::bytea, 'hub');

-- Download
SELECT nats_get_file('store', 'file.txt');

-- Delete
SELECT nats_delete_file('store', 'file.txt');

-- Metadata and listing
SELECT * FROM nats_get_file_info('store', 'file.txt');
SELECT * FROM nats_get_file_list('store');
SELECT * FROM nats_get_file_list('store', 'hub');

SELECT * FROM litmus_pgnats_version();     -- extension version
SELECT * FROM nats_get_server_info();      -- NATS server connection info

-- enable the extension in the database you want to implement your use case
CREATE EXTENSION IF NOT EXISTS litmus_pgnats;

-- setup the Foreign Server connection
CREATE SERVER nats_fdw_server_test FOREIGN DATA WRAPPER litmus_pgnats_fdw OPTIONS (	
	-- nats-external is the primary NATS server (JetStream domain: hub).
	host '10.30.50.1',

	-- TCP port for NATS connections
	port '4222',

	-- Internal command buffer size in messages
	capacity '128',

	-- pgnats NATS username (for LE Access Token leave set to empty string)
	username '',

	-- Litmus Edge Access Account token
	password '<Access Account token>');
	
-- relaod config
SELECT litmus_pgnats_reload_conf()

-- verify connection
SELECT * FROM nats_get_server_info();

-- create table to store LE payloads as simple string
CREATE TABLE IF NOT EXISTS litmus_pgnats.test_messages (
	received_at timestamptz DEFAULT now(),
	payload     text
);

-- create a function used by the subscriber to process the incoming payload
CREATE OR REPLACE FUNCTION litmus_pgnats.test_capture(msg bytea)
RETURNS void LANGUAGE sql AS $$
	INSERT INTO litmus_pgnats.test_messages(payload) VALUES (convert_from(msg, 'UTF8'));
$$;

-- Subscribe to topic from Litmus Edge
SELECT nats_subscribe('devicehub.alias.<mydevice>.<mytag>', 'litmus_pgnats.test_capture'::regproc); 

-- verify data are captured
SELECT * FROM litmus_pgnats.test_messages ORDER BY received_at DESC LIMIT 10;

-- cleanup
SELECT nats_unsubscribe('devicehub.alias.<mydevice>.<mytag>', 'litmus_pgnats.test_capture'::regproc);