Simple Queue system for Postgres using pg_cron & pg_net

# supa_queue Extension for Postgres

## Overview

The `supa_queue` extension is a simple but robust message queue system designed to work seamlessly with Postgres. It allows you to manage and process asynchronous jobs efficiently within your Postgres database.
You can read the blog post explaining this system here: [Building a Queue System with Supabase and PostgreSQL](https://blog.mansueli.com/building-a-queue-system-with-supabase-and-postgresql)

## Features

- Simple message queue system.
- Support for various HTTP methods (GET, POST, DELETE).
- Retry mechanism for failed jobs.
- Concurrent processing using multiple workers.
- Integration with external services via HTTP requests.
- Flexible scheduling with `pg_cron` for optimal job management.

## Installation

To install the `supa_queue` extension, follow these steps:

1. Make sure you have Postgres installed and running.

1. Install the required extensions `pg_cron` and `pg_net` if not already installed.

1. Install [DB.DEV](https://database.dev/installer):
```sql
create extension if not exists http with schema extensions;
create extension if not exists pg_tle;
select pgtle.uninstall_extension_if_exists('supabase-dbdev');
drop extension if exists "supabase-dbdev";
select
    pgtle.install_extension(
        'supabase-dbdev',
        resp.contents ->> 'version',
        'PostgreSQL package manager',
        resp.contents ->> 'sql'
    )
from http(
    (
        'GET',
        'https://api.database.dev/rest/v1/'
        || 'package_versions?select=sql,version'
        || '&package_name=eq.supabase-dbdev'
        || '&order=version.desc'
        || '&limit=1',
        array[
            ('apiKey', 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJzdXBhYmFzZSIsInJlZiI6InhtdXB0cHBsZnZpaWZyYndtbXR2Iiwicm9sZSI6ImFub24iLCJpYXQiOjE2ODAxMDczNzIsImV4cCI6MTk5NTY4MzM3Mn0.z2CN0mvO2No8wSi46Gw59DFGCTJrzM0AQKsu_5k134s')::http_header
        ],
        null,
        null
    )
) x,
lateral (
    select
        ((row_to_json(x) -> 'content') #>> '{}')::json -> 0
) resp(contents);
create extension "supabase-dbdev";
select dbdev.install('supabase-dbdev');
drop extension if exists "supabase-dbdev";
create extension "supabase-dbdev";
```

1. Install supa_queue with the following code:

```sql
select dbdev.install('mansueli-supa_queue');
create extension "mansueli-supa_queue"
    version '1.0.4';
```

You can also install it in a different `schema` with:

```sql
create schema supa_queue;
select dbdev.install('mansueli-supa_queue');
create extension "mansueli-supa_queue"
    schema supa_queue
    version '1.0.4';
```

## Usage

### Inserting Jobs

To add a job to the queue, insert a new record into the `job_queue` table. Specify the HTTP verb (GET, POST, DELETE), payload, and other relevant information. The job will be processed asynchronously.

```sql
INSERT INTO job_queue (http_verb, payload, url_path) VALUES ('GET', '{"key": "value"}', '/api/resource');
```

### Processing Jobs

Jobs are processed automatically using the provided functions. The `process_job()` trigger function processes newly inserted jobs. The `process_current_jobs_if_unlocked()` function assigns jobs to available workers for execution.

### Retrying Failed Jobs

Failed jobs are automatically retried with the `retry_failed_jobs()` function, increasing job reliability. Jobs with a status of 'failed' and within the retry limit will be retried.

## Configuration

You can configure various aspects of the `supa_queue` extension by modifying the provided SQL functions and cron schedules to suit your specific use case.

Note that you'll need to set these values in Vault:

 - `service_role` key you can get this in the [dashboard](https://supabase.com/dashboard/project/_/settings/api).
 - `consumer_function` this is the URL of the Edge Function that will consume the tasks.
## License

This extension is provided under the [license](https://github.com/mansueli/supa_queue/blob/master/LICENSE) included in the repository.

## Contributing

If you'd like to contribute to this project or report issues, please visit the [GitHub repository](https://github.com/mansueli/supa_queue) for more information.


In database version control for postgres functions

# Function VC (Version Control)

Function VC is a set of functions that allow you to version control your Postgres functions in the database.

This is a Trusted language extension to allow for an easy install of the version control of Postgres functions that was published on this blog post: [Easy Deployment and Rollback of PostgreSQL Functions with Supabase](https://blog.mansueli.com/streamlining-postgresql-function-management-with-supabase).

## Installing FunctionVC

### Install Database.dev
<details>

<summary>(if you don't have it)</summary>

```sql
/*---------------------
---- install dbdev ----
----------------------
Requires:
  - pg_tle: https://github.com/aws/pg_tle
  - pgsql-http: https://github.com/pramsey/pgsql-http
*/
create extension if not exists http with schema extensions;
create extension if not exists pg_tle;
drop extension if exists "supabase-dbdev";
select pgtle.uninstall_extension_if_exists('supabase-dbdev');
select
    pgtle.install_extension(
        'supabase-dbdev',
        resp.contents ->> 'version',
        'PostgreSQL package manager',
        resp.contents ->> 'sql'
    )
from http(
    (
        'GET',
        'https://api.database.dev/rest/v1/'
        || 'package_versions?select=sql,version'
        || '&package_name=eq.supabase-dbdev'
        || '&order=version.desc'
        || '&limit=1',
        array[
            (
                'apiKey',
                'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJp'
                || 'c3MiOiJzdXBhYmFzZSIsInJlZiI6InhtdXB0cHBsZnZpaWZyY'
                || 'ndtbXR2Iiwicm9sZSI6ImFub24iLCJpYXQiOjE2ODAxMDczNzI'
                || 'sImV4cCI6MTk5NTY4MzM3Mn0.z2CN0mvO2No8wSi46Gw59DFGCTJ'
                || 'rzM0AQKsu_5k134s'
            )::http_header
        ],
        null,
        null
    )
) x,
lateral (
    select
        ((row_to_json(x) -> 'content') #>> '{}')::json -> 0
) resp(contents);
create extension "supabase-dbdev";
select dbdev.install('supabase-dbdev');
drop extension if exists "supabase-dbdev";
create extension "supabase-dbdev";
```

</details>

### Installing Function VC

```sql
select dbdev.install('mansueli-function_vc');
create extension "mansueli-function_vc"
    version '1.0.1';
```

## Creating a function:

```sql
SELECT create_function_from_source(
$$
-- Note that you can just paste the function below:
CREATE OR REPLACE FUNCTION public.convert_to_uuid(input_value text)
 RETURNS uuid
AS $function$
DECLARE
  hash_hex text;
BEGIN
  -- Return null if input_value is null or an empty string
  IF input_value IS NULL OR NULLIF(input_value, '') IS NULL THEN
    RETURN NULL;
  END IF;
  hash_hex := substring(encode(digest(input_value::bytea, 'sha512'), 'hex'), 1, 36);
  RETURN (left(hash_hex, 8) || '-' || right(hash_hex, 4) || '-4' || right(hash_hex, 3) || '-a' || right(hash_hex, 3) || '-' || right(hash_hex, 12))::uuid;
END;
$function$
LANGUAGE plpgsql
IMMUTABLE
SECURITY DEFINER;
-- End of the function above
$$
);
```

Rolling back the latest version of a function:

```sql

SELECT rollback_function('convert_to_uuid');

```

Rolling back a specific version of a function:

```sql

SELECT rollback_function('convert_to_uuid', 2);

```

View the function history and versions

```sql

SELECT * from archive.function_history
  WHERE schema_name = 'public' 
  and function_name ='convert_to_uuid' ;

```

Check the current definition of a function

```sql

SELECT pg_get_functiondef((SELECT oid FROM pg_proc
  WHERE proname = 'convert_to_uuid'));

```

Storing all existing functions in the database:

```sql
SELECT archive.setup_function_history();
```

## Thanks to our contributors:
- [@imor](https://github.com/imor/)
- [@idea-garage](https://github.com/idea-garage/)


An extension to help with RLS policies within the database context

## RLS Helpers

This is an extension with convenience functions to help test and simulate RLS policies within the database.

You can use use it to get RLS context within direct database access or with Supavisor:

## Usage

You can simulate log in as an user with the following:

### Login as user:

```sql
BEGIN;
CALL auth.login_as_user('rodrigo@contoso.com');
SELECT * FROM profiles;
COMMIT;
```

### Anon access:
```sql
BEGIN;
CALL auth.login_as_anon();
SELECT * FROM profiles;
COMMIT;
```


### Logout:

```sql
BEGIN;
CALL auth.login_as_user('rodrigo@contoso.com');
SELECT * FROM profiles;
CALL auth.logout();
-- You are back to postgres role here:
SELECT * FROM profiles;
COMMIT;
```

A very powerful extension to use webhooks and interact with Supabase Edge Functions

# pgwebhook

`pgwebhook` is a PostgreSQL extension designed to facilitate the creation and management of webhooks directly from your database. It works seamlessly with the Supabase platform and leverages the `pghttp` extension for making HTTP requests. This extension is ideal for developers looking to integrate real-time data changes in PostgreSQL with external services and applications through webhooks.

## Prerequisites

- PostgreSQL 14 or newer
- `pghttp` extension (version 1.5.0 or newer recommended)

## Installation

For Supabase users or those with TLE (Trusted Language Extensions) installed, you can easily install `pgwebhook` using the following SQL commands:

```sql
SELECT dbdev.install('mansueli@pgwebhook');
CREATE EXTENSION "mansueli@pgwebhook" VERSION '0.1.1';
```

### Manual Installation

If you're not using Supabase or TLE, you can manually install `pgwebhook` by running the SQL script for the latest version (e.g., `pgwebhook--0.1.1.sql`). Ensure you have the `pgsql-http` extension installed on your PostgreSQL database. For detailed instructions, visit the [pgsql-http GitHub repository](https://github.com/pramsey/pgsql-http).

## Usage

This extension supports two main usage scenarios: direct calls to external services and webhook triggers.

### Direct Usage

#### Calling the Edge Functions Within the Database (e.g., pg_cron or through an RPC)

We believe that while this extension offers a lot of options, most projects will benefit from a wrapper that sets defaults around it. For example, let's say OpenAI restricts calls to allowed regions, or you need to keep your Edge Function calls within a few subregions for GDPR compliance.

```sql
-- Create wrapper function in the public schema
CREATE OR REPLACE FUNCTION public.euro_edge (func TEXT, data JSONB) RETURNS JSONB LANGUAGE plpgsql
AS $function$
DECLARE
    custom_headers JSONB;
    allowed_regions TEXT[] := ARRAY['eu-west-1', 'eu-west-2', 'eu-west-3', 'eu-north-1', 'eu-central-1'];
BEGIN
    -- Set headers with anon key and Content-Type
    custom_headers := jsonb_build_object('Authorization', vault.get_anon_key(bearer := true), 'Content-Type', 'application/json', 'x-region', allowed_regions);
    -- Call edge_wrapper function with default values
    RETURN hook.edge_wrapper(url := ('https://supanacho.supabase.co/functions/v1/' || func), headers := custom_headers, payload := data, max_retries := 5, allowed_regions := allowed_regions);
END;
$function$;
```

Then you can call this from the supabase-js client like:

```js
const { data, error } = await supabase.rpc('euro_edge', {
    func: 'hello-world',
    data: JSON.stringify({ name: 'John Doe' })
});
```

### Webhooks

#### Creating Webhook Triggers

To create a webhook trigger for a table, use the `hook.webhook_trigger` function as follows:

```sql
CREATE TRIGGER your_trigger_name
AFTER INSERT OR UPDATE OR DELETE ON your_table
FOR EACH ROW EXECUTE FUNCTION hook.webhook_trigger(
    'https://your-webhook-url.com',
    'POST',
    '{"Content-Type": "application/json"}',
    '{}',
    5000,
    'your_custom_handler'
);
```

#### Creating Edge Webhook Triggers

To create an edge webhook trigger that retries requests and handles regional headers, use the `hook.edgehook_trigger` function:

```sql
CREATE TRIGGER your_edge_trigger_name
AFTER INSERT OR UPDATE OR DELETE ON your_table
FOR EACH ROW EXECUTE FUNCTION hook.edgehook_trigger(
    'https://your-webhook-url.com',
    'POST',
    '{"Content-Type": "application/json"}',
    '{}',
    5000,
    ARRAY['region1', 'region2']
);
```

### Parameters

- **url**: The URL to send the webhook request to.
- **method**: HTTP method to use (`GET`, `POST`, `PUT`, `PATCH`, `DELETE`, `HEAD`).
- **headers**: JSON object containing the headers for the HTTP request.
- **params**: JSON object containing the query parameters for the HTTP request.
- **timeout_ms**: Timeout in milliseconds for the HTTP request.
- **max_retries**: Maximum number of retries for the HTTP request in case of failures.
- **allowed_regions**: Array of allowed regions for edge triggers.
- **custom_handler**: Optional custom handler function for constructing the payload.

### List of Allowed Regions

- ap-northeast-1
- ap-northeast-2
- ap-south-1
- ap-southeast-1
- ap-southeast-2
- ca-central-1
- eu-central-1
- eu-west-1
- eu-west-2
- eu-west-3
- sa-east-1
- us-east-1
- us-west-1
- us-west-2

### Using Triggers on a Table

#### Create a Trigger Using a PL/pgSQL Custom Handler

```sql
CREATE OR REPLACE FUNCTION hook.custom_handler_function(payload jsonb)
RETURNS jsonb AS $$
DECLARE
    new_payload jsonb;
BEGIN
    -- Modify the payload as needed
    -- Example: Add a new field to the payload
    new_payload := payload || jsonb_build_object('additional_info', 'This is extra info');

    -- Example: Remove a sensitive field
    new_payload := new_payload - 'sensitive_field';

    -- Return the modified payload
    RETURN new_payload;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER your_trigger_name
AFTER INSERT OR UPDATE OR DELETE ON your_table
FOR EACH ROW EXECUTE FUNCTION hook.webhook_trigger(
    'https://your-webhook-url.com',
    'POST',
    '{"Content-Type": "application/json"}',
    '{}',
    5000,
    'hook.custom_handler_function'
);
```

#### Create a Trigger Using a PLv8 Custom Handler

```sql
CREATE OR REPLACE FUNCTION hook.custom_handler_function_js(payload jsonb)
RETURNS jsonb AS $$
var newPayload = payload;

// Modify the payload: Add new field and remove a sensitive field
newPayload.additional_info = 'This is extra info';
delete newPayload.sensitive_field;

return newPayload;
$$ LANGUAGE plv8;

CREATE TRIGGER your_trigger_name
AFTER INSERT OR UPDATE OR DELETE ON your_table
FOR EACH ROW EXECUTE FUNCTION hook.webhook_trigger(
    'https://your-webhook-url.com',
    'POST',
    '{"Content-Type": "application/json"}',
    '{}',
    5000,
    'hook.custom_handler_function_js'
);
```

## License

This code is licensed under the [Apache License 2.0](https://github.com/mansueli/tle/blob/main/LICENSE).


mansueli

Packages