SELECT pg_catalog.set_config('search_path', '', false);

CREATE SCHEMA graphile_worker;

ALTER SCHEMA graphile_worker OWNER TO graphile_worker_role;

CREATE EXTENSION IF NOT EXISTS pgcrypto WITH SCHEMA public;

COMMENT ON EXTENSION pgcrypto IS 'cryptographic functions';

CREATE TABLE graphile_worker.jobs (

id bigint NOT NULL,

queue_name text DEFAULT (public.gen_random_uuid())::text,

task_identifier text NOT NULL,

payload json DEFAULT '{}'::json NOT NULL,

priority integer DEFAULT 0 NOT NULL,

run_at timestamp with time zone DEFAULT now() NOT NULL,

attempts integer DEFAULT 0 NOT NULL,

max_attempts integer DEFAULT 25 NOT NULL,

last_error text,

created_at timestamp with time zone DEFAULT now() NOT NULL,

updated_at timestamp with time zone DEFAULT now() NOT NULL,

key text,

locked_at timestamp with time zone,

locked_by text,

revision integer DEFAULT 0 NOT NULL,

CONSTRAINT jobs_key_check CHECK ((length(key) > 0))

);

ALTER TABLE graphile_worker.jobs OWNER TO graphile_worker_role;

CREATE FUNCTION graphile_worker.add_job(identifier text, payload json DEFAULT NULL::json, queue_name text DEFAULT NULL::text, run_at timestamp with time zone DEFAULT NULL::timestamp with time zone, max_attempts integer DEFAULT NULL::integer, job_key text DEFAULT NULL::text, priority integer DEFAULT NULL::integer) RETURNS graphile_worker.jobs

LANGUAGE plpgsql

AS $$

declare

v_job "graphile_worker".jobs;

-- Apply rationality checks

if length(identifier) > 128 then

raise exception 'Task identifier is too long (max length: 128).' using errcode = 'GWBID';

end if;

if queue_name is not null and length(queue_name) > 128 then

raise exception 'Job queue name is too long (max length: 128).' using errcode = 'GWBQN';

end if;

if job_key is not null and length(job_key) > 512 then

raise exception 'Job key is too long (max length: 512).' using errcode = 'GWBJK';

end if;

if max_attempts < 1 then

raise exception 'Job maximum attempts must be at least 1' using errcode = 'GWBMA';

end if;

if job_key is not null then

-- Upsert job

insert into "graphile_worker".jobs (

task_identifier,

payload,

queue_name,

run_at,

max_attempts,

key,

priority

)

values(

identifier,

coalesce(payload, '{}'::json),

queue_name,

coalesce(run_at, now()),

coalesce(max_attempts, 25),

job_key,

coalesce(priority, 0)

)

on conflict (key) do update set

task_identifier=excluded.task_identifier,

payload=excluded.payload,

queue_name=excluded.queue_name,

max_attempts=excluded.max_attempts,

run_at=excluded.run_at,

priority=excluded.priority,

revision=jobs.revision + 1,

-- always reset error/retry state

attempts=0,

last_error=null

where jobs.locked_at is null

returning *

into v_job;

-- If upsert succeeded (insert or update), return early

if not (v_job is null) then

return v_job;

end if;

-- Upsert failed -> there must be an existing job that is locked. Remove

-- existing key to allow a new one to be inserted, and prevent any

-- subsequent retries by bumping attempts to the max allowed.

update "graphile_worker".jobs

set

key = null,

attempts = jobs.max_attempts

where key = job_key;

end if;

-- insert the new job. Assume no conflicts due to the update above

insert into "graphile_worker".jobs(

task_identifier,

payload,

queue_name,

run_at,

max_attempts,

key,

priority

)

values(

identifier,

coalesce(payload, '{}'::json),

queue_name,

coalesce(run_at, now()),

coalesce(max_attempts, 25),

job_key,

coalesce(priority, 0)

)

returning *

into v_job;

return v_job;

end;

$$;

ALTER FUNCTION graphile_worker.add_job(identifier text, payload json, queue_name text, run_at timestamp with time zone, max_attempts integer, job_key text, priority integer) OWNER TO graphile_worker_role;

CREATE FUNCTION graphile_worker.complete_job(worker_id text, job_id bigint) RETURNS graphile_worker.jobs

LANGUAGE plpgsql

AS $$

declare

v_row "graphile_worker".jobs;

delete from "graphile_worker".jobs

where id = job_id

returning * into v_row;

if v_row.queue_name is not null then

update "graphile_worker".job_queues

set locked_by = null, locked_at = null

where queue_name = v_row.queue_name and locked_by = worker_id;

end if;

return v_row;

end;

$$;

ALTER FUNCTION graphile_worker.complete_job(worker_id text, job_id bigint) OWNER TO graphile_worker_role;

CREATE FUNCTION graphile_worker.complete_jobs(job_ids bigint[]) RETURNS SETOF graphile_worker.jobs

LANGUAGE sql

AS $$

delete from "graphile_worker".jobs

where id = any(job_ids)

and (

locked_by is null

or

locked_at < NOW() - interval '4 hours'

)

returning *;

$$;

ALTER FUNCTION graphile_worker.complete_jobs(job_ids bigint[]) OWNER TO graphile_worker_role;

CREATE FUNCTION graphile_worker.fail_job(worker_id text, job_id bigint, error_message text) RETURNS graphile_worker.jobs

LANGUAGE plpgsql STRICT

AS $$

declare

v_row "graphile_worker".jobs;

update "graphile_worker".jobs

set

last_error = error_message,

run_at = greatest(now(), run_at) + (exp(least(attempts, 10))::text || ' seconds')::interval,

locked_by = null,

locked_at = null

where id = job_id and locked_by = worker_id

returning * into v_row;

if v_row.queue_name is not null then

update "graphile_worker".job_queues

set locked_by = null, locked_at = null

where queue_name = v_row.queue_name and locked_by = worker_id;

end if;

return v_row;

end;

$$;

ALTER FUNCTION graphile_worker.fail_job(worker_id text, job_id bigint, error_message text) OWNER TO graphile_worker_role;

CREATE FUNCTION graphile_worker.get_job(worker_id text, task_identifiers text[] DEFAULT NULL::text[], job_expiry interval DEFAULT '04:00:00'::interval) RETURNS graphile_worker.jobs

LANGUAGE plpgsql

AS $$

declare

v_job_id bigint;

v_queue_name text;

v_row "graphile_worker".jobs;

v_now timestamptz = now();

if worker_id is null or length(worker_id) < 10 then

raise exception 'invalid worker id';

end if;

select jobs.queue_name, jobs.id into v_queue_name, v_job_id

from "graphile_worker".jobs

where (jobs.locked_at is null or jobs.locked_at < (v_now - job_expiry))

and (

jobs.queue_name is null

or

exists (

select 1

from "graphile_worker".job_queues

where job_queues.queue_name = jobs.queue_name

and (job_queues.locked_at is null or job_queues.locked_at < (v_now - job_expiry))

for update

skip locked

)

)

and run_at <= v_now

and attempts < max_attempts

and (task_identifiers is null or task_identifier = any(task_identifiers))

order by priority asc, run_at asc, id asc

limit 1

for update

skip locked;

if v_job_id is null then

return null;

end if;

if v_queue_name is not null then

update "graphile_worker".job_queues

set

locked_by = worker_id,

locked_at = v_now

where job_queues.queue_name = v_queue_name;

end if;

update "graphile_worker".jobs

set

attempts = attempts + 1,

locked_by = worker_id,

locked_at = v_now

where id = v_job_id

returning * into v_row;

return v_row;

end;

$$;

ALTER FUNCTION graphile_worker.get_job(worker_id text, task_identifiers text[], job_expiry interval) OWNER TO graphile_worker_role;

CREATE FUNCTION graphile_worker.jobs__decrease_job_queue_count() RETURNS trigger

LANGUAGE plpgsql

AS $$

declare

v_new_job_count int;

update "graphile_worker".job_queues

set job_count = job_queues.job_count - 1

where queue_name = old.queue_name

returning job_count into v_new_job_count;

if v_new_job_count <= 0 then

delete from "graphile_worker".job_queues where queue_name = old.queue_name and job_count <= 0;

end if;

return old;

end;

$$;

ALTER FUNCTION graphile_worker.jobs__decrease_job_queue_count() OWNER TO graphile_worker_role;

CREATE FUNCTION graphile_worker.jobs__increase_job_queue_count() RETURNS trigger

LANGUAGE plpgsql

AS $$

insert into "graphile_worker".job_queues(queue_name, job_count)

values(new.queue_name, 1)

on conflict (queue_name)

do update

set job_count = job_queues.job_count + 1;

return new;

end;

$$;

ALTER FUNCTION graphile_worker.jobs__increase_job_queue_count() OWNER TO graphile_worker_role;

CREATE FUNCTION graphile_worker.permanently_fail_jobs(job_ids bigint[], error_message text DEFAULT NULL::text) RETURNS SETOF graphile_worker.jobs

LANGUAGE sql

AS $$

update "graphile_worker".jobs

set

last_error = coalesce(error_message, 'Manually marked as failed'),

attempts = max_attempts

where id = any(job_ids)

and (

locked_by is null

or

locked_at < NOW() - interval '4 hours'

)

returning *;

$$;

ALTER FUNCTION graphile_worker.permanently_fail_jobs(job_ids bigint[], error_message text) OWNER TO graphile_worker_role;

CREATE FUNCTION graphile_worker.remove_job(job_key text) RETURNS graphile_worker.jobs

LANGUAGE sql STRICT

AS $$

delete from "graphile_worker".jobs

where key = job_key

and locked_at is null

returning *;

$$;

ALTER FUNCTION graphile_worker.remove_job(job_key text) OWNER TO graphile_worker_role;

CREATE FUNCTION graphile_worker.reschedule_jobs(job_ids bigint[], run_at timestamp with time zone DEFAULT NULL::timestamp with time zone, priority integer DEFAULT NULL::integer, attempts integer DEFAULT NULL::integer, max_attempts integer DEFAULT NULL::integer) RETURNS SETOF graphile_worker.jobs

LANGUAGE sql

AS $$

update "graphile_worker".jobs

set

run_at = coalesce(reschedule_jobs.run_at, jobs.run_at),

priority = coalesce(reschedule_jobs.priority, jobs.priority),

attempts = coalesce(reschedule_jobs.attempts, jobs.attempts),

max_attempts = coalesce(reschedule_jobs.max_attempts, jobs.max_attempts)

where id = any(job_ids)

and (

locked_by is null

or

locked_at < NOW() - interval '4 hours'

)

returning *;

$$;

ALTER FUNCTION graphile_worker.reschedule_jobs(job_ids bigint[], run_at timestamp with time zone, priority integer, attempts integer, max_attempts integer) OWNER TO graphile_worker_role;

CREATE FUNCTION graphile_worker.tg__update_timestamp() RETURNS trigger

LANGUAGE plpgsql

AS $$

new.updated_at = greatest(now(), old.updated_at + interval '1 millisecond');

return new;

end;

$$;

ALTER FUNCTION graphile_worker.tg__update_timestamp() OWNER TO graphile_worker_role;

CREATE FUNCTION graphile_worker.tg_jobs__notify_new_jobs() RETURNS trigger

LANGUAGE plpgsql

AS $$

perform pg_notify('jobs:insert', '');

return new;

end;

$$;

ALTER FUNCTION graphile_worker.tg_jobs__notify_new_jobs() OWNER TO graphile_worker_role;

CREATE TABLE graphile_worker.job_queues (

queue_name text NOT NULL,

job_count integer NOT NULL,

locked_at timestamp with time zone,

locked_by text

);

ALTER TABLE graphile_worker.job_queues OWNER TO graphile_worker_role;

CREATE SEQUENCE graphile_worker.jobs_id_seq

START WITH 1

INCREMENT BY 1

NO MINVALUE

NO MAXVALUE

CACHE 1;

ALTER TABLE graphile_worker.jobs_id_seq OWNER TO graphile_worker_role;

ALTER SEQUENCE graphile_worker.jobs_id_seq OWNED BY graphile_worker.jobs.id;

CREATE TABLE graphile_worker.migrations (

id integer NOT NULL,

ts timestamp with time zone DEFAULT now() NOT NULL

);

ALTER TABLE graphile_worker.migrations OWNER TO graphile_worker_role;

ALTER TABLE ONLY graphile_worker.jobs ALTER COLUMN id SET DEFAULT nextval('graphile_worker.jobs_id_seq'::regclass);

ALTER TABLE ONLY graphile_worker.job_queues

ADD CONSTRAINT job_queues_pkey PRIMARY KEY (queue_name);

ALTER TABLE ONLY graphile_worker.jobs

ADD CONSTRAINT jobs_key_key UNIQUE (key);

ALTER TABLE ONLY graphile_worker.jobs

ADD CONSTRAINT jobs_pkey PRIMARY KEY (id);

ALTER TABLE ONLY graphile_worker.migrations

ADD CONSTRAINT migrations_pkey PRIMARY KEY (id);

CREATE INDEX jobs_priority_run_at_id_idx ON graphile_worker.jobs USING btree (priority, run_at, id);

CREATE TRIGGER _100_timestamps BEFORE UPDATE ON graphile_worker.jobs FOR EACH ROW EXECUTE PROCEDURE graphile_worker.tg__update_timestamp();

CREATE TRIGGER _500_decrease_job_queue_count AFTER DELETE ON graphile_worker.jobs FOR EACH ROW WHEN ((old.queue_name IS NOT NULL)) EXECUTE PROCEDURE graphile_worker.jobs__decrease_job_queue_count();

CREATE TRIGGER _500_decrease_job_queue_count_update AFTER UPDATE OF queue_name ON graphile_worker.jobs FOR EACH ROW WHEN (((new.queue_name IS DISTINCT FROM old.queue_name) AND (old.queue_name IS NOT NULL))) EXECUTE PROCEDURE graphile_worker.jobs__decrease_job_queue_count();

CREATE TRIGGER _500_increase_job_queue_count AFTER INSERT ON graphile_worker.jobs FOR EACH ROW WHEN ((new.queue_name IS NOT NULL)) EXECUTE PROCEDURE graphile_worker.jobs__increase_job_queue_count();

CREATE TRIGGER _500_increase_job_queue_count_update AFTER UPDATE OF queue_name ON graphile_worker.jobs FOR EACH ROW WHEN (((new.queue_name IS DISTINCT FROM old.queue_name) AND (new.queue_name IS NOT NULL))) EXECUTE PROCEDURE graphile_worker.jobs__increase_job_queue_count();

CREATE TRIGGER _900_notify_worker AFTER INSERT ON graphile_worker.jobs FOR EACH STATEMENT EXECUTE PROCEDURE graphile_worker.tg_jobs__notify_new_jobs();

ALTER TABLE graphile_worker.job_queues ENABLE ROW LEVEL SECURITY;

ALTER TABLE graphile_worker.jobs ENABLE ROW LEVEL SECURITY;