BEGIN;

CREATE EXTENSION IF NOT EXISTS pgcrypto SCHEMA public CASCADE;

CREATE TYPE task_status_code_value AS ENUM (

'pending', -- waiting for an actor to perform it (new or being retried)

'in-progress', -- task is being performed by an actor

'aborted', -- an actor has aborted this task with message, will not be reissued

'completed' -- an actor has completed this task with message, will not reissued

);

CREATE TABLE IF NOT EXISTS tasks (

id VARCHAR NOT NULL PRIMARY KEY, -- nanoid

action VARCHAR NOT NULL, -- name (type) of action to perform

body TEXT, -- data used by action

status TEXT, -- status text defined by action, visible to action

status_code task_status_code_value NOT NULL DEFAULT 'pending', -- internal status code, used by parade to issue tasks

num_tries INTEGER NOT NULL DEFAULT 0, -- number of attempts actors have made on this task

num_failures INTEGER NOT NULL DEFAULT 0, -- number of those attempts that failed

max_tries INTEGER,

total_dependencies INTEGER, -- number of tasks that must signal this task

num_signals INTEGER NOT NULL DEFAULT 0, -- number of tasks that have already signalled this task

actor_id VARCHAR, -- ID of performing actor if in-progress

action_deadline TIMESTAMPTZ, -- offer this task to other actors once action_deadline has elapsed

performance_token UUID,

-- BUG(ariels): add REFERENCES dependency to each of the to_signal_after

-- tasks. Or at least add triggers that perform ON DELETE

-- CASCADE.

to_signal_after VARCHAR ARRAY, -- IDs to signal after performing this task

notify_channel_after VARCHAR -- (if non-NULL) name of a channel to NOTIFY when this task ends

);

CREATE OR REPLACE FUNCTION can_allocate_task(id VARCHAR, code task_status_code_value, deadline TIMESTAMPTZ, num_signals INTEGER, total_dependencies INTEGER)

RETURNS BOOLEAN

LANGUAGE sql IMMUTABLE AS $$

SELECT (code = 'pending' OR (code = 'in-progress' AND deadline < NOW())) AND

(total_dependencies IS NULL OR num_signals = total_dependencies)

$$;

CREATE OR REPLACE FUNCTION own_tasks(

max_tasks INTEGER, actions VARCHAR ARRAY, owner_id VARCHAR, max_duration INTERVAL

)

RETURNS TABLE(task_id VARCHAR, token UUID, num_failures INTEGER, action VARCHAR, body TEXT)

LANGUAGE sql VOLATILE AS $$

UPDATE tasks

SET actor_id = owner_id,

status_code = 'in-progress',

num_tries = num_tries + 1,

performance_token = public.gen_random_uuid(),

action_deadline = NOW() + max_duration -- NULL if max_duration IS NULL

WHERE id IN (

SELECT id

FROM tasks

WHERE can_allocate_task(id, status_code, action_deadline, num_signals, total_dependencies) AND

action = ANY(actions) AND

(max_tries IS NULL OR num_tries < max_tries)

-- maybe: AND not_before <= NOW()

-- maybe: ORDER BY priority (eventually)

ORDER BY random()

FOR UPDATE SKIP LOCKED

LIMIT max_tasks)

RETURNING id, performance_token, num_failures, action, body

$$;

CREATE OR REPLACE FUNCTION extend_task_deadline(

task_id VARCHAR, token UUID, max_duration INTERVAL

) RETURNS BOOLEAN

LANGUAGE sql VOLATILE AS $$

UPDATE tasks

SET action_deadline = NOW() + max_duration

WHERE id = task_id AND performance_token = token

RETURNING true;

$$;

CREATE OR REPLACE FUNCTION no_more_tries(r tasks)

RETURNS BOOLEAN LANGUAGE sql IMMUTABLE AS $$

SELECT $1.num_tries >= $1.max_tries

$$;

CREATE OR REPLACE FUNCTION return_task(

task_id VARCHAR, token UUID, result_status TEXT, result_status_code task_status_code_value

) RETURNS INTEGER

LANGUAGE plpgsql AS $$

DECLARE

num_updated INTEGER;

channel VARCHAR;

to_signal VARCHAR ARRAY;

CASE result_status_code

WHEN 'aborted', 'completed' THEN

UPDATE tasks INTO channel, to_signal

SET status = result_status,

status_code = result_status_code,

actor_id = NULL,

performance_token = NULL

WHERE id = task_id AND performance_token = token

RETURNING notify_channel_after, to_signal_after;

WHEN 'pending' THEN

UPDATE tasks INTO channel, to_signal

SET status = result_status,

status_code = (CASE WHEN no_more_tries(tasks) THEN 'aborted' ELSE 'pending' END)::task_status_code_value,

actor_id = NULL,

performance_token = NULL

WHERE id = task_id AND performance_token = token

RETURNING (CASE WHEN no_more_tries(tasks) THEN notify_channel_after ELSE NULL END),

(CASE WHEN no_more_tries(tasks) THEN to_signal_after ELSE NULL END);

ELSE

RAISE EXCEPTION 'cannot return task to status %', result_status;

END CASE;

GET DIAGNOSTICS num_updated := ROW_COUNT;

UPDATE tasks

SET num_signals = num_signals+1,

num_failures = num_failures + CASE WHEN result_status_code = 'aborted'::task_status_code_value THEN 1 ELSE 0 END

WHERE id = ANY(to_signal);

IF channel IS NOT NULL THEN

PERFORM pg_notify(channel, NULL);

END IF;

RETURN num_updated;

END;

$$;

CREATE OR REPLACE FUNCTION remove_task_dependencies(task_id VARCHAR)

RETURNS SETOF VARCHAR

LANGUAGE sql VOLATILE AS $$

WITH updates AS (

SELECT UNNEST(to_signal_after) effect_id,

(CASE WHEN status_code IN ('aborted', 'completed') THEN 0 ELSE 1 END) delta

FROM tasks

WHERE id=task_id),

signalled_ids AS (

UPDATE tasks

SET total_dependencies = total_dependencies-updates.delta

FROM updates

WHERE id=updates.effect_id

RETURNING (CASE WHEN total_dependencies = num_signals THEN tasks.id ELSE NULL END) id

)

SELECT id FROM signalled_ids WHERE id IS NOT NULL;

$$;

CREATE TYPE tasks_recurse_value AS ENUM ('new', 'in-progress', 'done');

CREATE OR REPLACE FUNCTION delete_tasks(task_id_name TEXT) RETURNS VOID LANGUAGE plpgsql AS $$

DECLARE

total_num_updated INTEGER;

num_updated INTEGER;

row_count INTEGER;

LOOP

EXECUTE format($Q$

UPDATE %1$I SET mark='in-progress' WHERE mark='new'

$Q$, task_id_name);

EXECUTE format($Q$

WITH new_to_delete AS (

SELECT remove_task_dependencies(id) id FROM %1$I WHERE mark='in-progress'

)

INSERT INTO %1$I (SELECT id, 'new' mark FROM new_to_delete)

$Q$, task_id_name);

GET DIAGNOSTICS row_count = ROW_COUNT;

EXIT WHEN row_count=0;

EXECUTE format($Q$

UPDATE %1$I SET mark='done' WHERE mark='in-progress'

$Q$, task_id_name);

END LOOP;

EXECUTE format($Q$

DELETE FROM tasks WHERE id IN (SELECT id FROM %1$I)

$Q$, task_id_name);

END;

$$;

END;

BEGIN;

DROP FUNCTION IF EXISTS delete_tasks;

DROP TYPE IF EXISTS tasks_recurse_value;

DROP FUNCTION IF EXISTS remove_task_dependencies;

DROP FUNCTION IF EXISTS return_task;

DROP FUNCTION IF EXISTS no_more_tries;

DROP FUNCTION IF EXISTS extend_task_deadline;

DROP FUNCTION IF EXISTS own_tasks;

DROP FUNCTION IF EXISTS can_allocate_task;

DROP TABLE IF EXISTS tasks;

DROP TYPE IF EXISTS task_status_code_value;

END;

github.com/jackc/pgtype v1.6.2 // indirect

github.com/lib/pq v1.9.0 // indirect