0053.yml

version: 53
description: Use task_queue_id instead of provisioner_id / worker_type in queues
migrationScript: |-
  begin
    lock table queue_worker_types;

    alter table queue_worker_types drop constraint queue_worker_types_pkey;
    alter table queue_worker_types rename to task_queues;
    alter table task_queues add column task_queue_id text;
    update task_queues set task_queue_id = provisioner_id || '/' || worker_type;
    alter table task_queues drop column provisioner_id, drop column worker_type;
    alter table task_queues alter column task_queue_id set not null;
    alter table task_queues add primary key (task_queue_id);

    lock table queue_workers;

    alter table queue_workers drop constraint queue_workers_pkey;
    alter table queue_workers add column task_queue_id text;
    update queue_workers set task_queue_id = provisioner_id || '/' || worker_type;
    alter table queue_workers drop column provisioner_id, drop column worker_type;
    alter table queue_workers alter column task_queue_id set not null;
    alter table queue_workers add primary key (task_queue_id, worker_group, worker_id);

    lock table queue_provisioners;

    revoke select, insert, update, delete on queue_provisioners from $db_user_prefix$_queue;
    drop table queue_provisioners;
  end
downgradeScript: |-
  begin
    lock table task_queues;

    alter table task_queues drop constraint task_queues_pkey;
    alter table task_queues add column provisioner_id text, add column worker_type text;
    update task_queues
    set
      provisioner_id = split_part(task_queues.task_queue_id, '/', 1),
      worker_type = split_part(task_queues.task_queue_id, '/', 2);
    alter table task_queues drop column task_queue_id;
    alter table task_queues alter column provisioner_id set not null;
    alter table task_queues alter column worker_type set not null;
    alter table task_queues add primary key (provisioner_id, worker_type);
    alter table task_queues rename to queue_worker_types;

    lock table queue_workers;

    alter table queue_workers drop constraint queue_workers_pkey;
    alter table queue_workers add column provisioner_id text, add column worker_type text;
    update queue_workers
    set
      provisioner_id = split_part(queue_workers.task_queue_id, '/', 1),
      worker_type = split_part(queue_workers.task_queue_id, '/', 2);
    alter table queue_workers drop column task_queue_id;
    alter table queue_workers alter column provisioner_id set not null;
    alter table queue_workers alter column worker_type set not null;
    alter table queue_workers add primary key (provisioner_id, worker_type, worker_group, worker_id);

    create table queue_provisioners
    as
      select
        provisioner_id,
        max(expires) as expires,
        max(last_date_active) as last_date_active,
        '' as description,
        'experimental' as stability,
        '[]'::jsonb as actions
      from queue_worker_types
      group by provisioner_id;

    alter table queue_provisioners add primary key (provisioner_id);
    alter table queue_provisioners
      alter column provisioner_id set not null,
      alter column expires set not null,
      alter column last_date_active set not null,
      alter column description set not null,
      alter column stability set not null,
      alter column actions set not null;

    grant select, insert, update, delete on queue_provisioners to $db_user_prefix$_queue;
  end
methods:
  create_queue_worker_type:
    description: |-
      Create a new queue worker type. Raises UNIQUE_VIOLATION if the worker type already exists.
    deprecated: true
    mode: write
    serviceName: queue
    args: provisioner_id_in text, worker_type_in text, expires_in timestamptz, last_date_active_in timestamptz, description_in text, stability_in text
    returns: uuid
    body: |-
      begin
        insert
          into task_queues (task_queue_id, expires, last_date_active, description, stability)
          values (
            provisioner_id_in || '/' || worker_type_in,
            expires_in,
            last_date_active_in,
            description_in, stability_in
          );
          return public.gen_random_uuid();
      end
  get_queue_worker_type:
    description: |-
      Get a non-expired queue worker type by provisioner_id and worker_type.
    deprecated: true
    mode: read
    serviceName: queue
    args: provisioner_id_in text, worker_type_in text, expires_in timestamptz
    returns: table(provisioner_id text, worker_type text, expires timestamptz, last_date_active timestamptz, description text, stability text, etag uuid)
    body: |-
      begin
        return query
        select
          split_part(task_queues.task_queue_id, '/', 1) as provisioner_id,
          split_part(task_queues.task_queue_id, '/', 2) as worker_type,
          task_queues.expires,
          task_queues.last_date_active,
          task_queues.description,
          task_queues.stability,
          public.gen_random_uuid()
        from task_queues
        where
          task_queues.task_queue_id = provisioner_id_in || '/' || worker_type_in and
          task_queues.expires > expires_in;
        end
  update_queue_worker_type:
    serviceName: queue
    description: |-
      Update a queue worker type's expires, last_date_active, description, and stability.
      All parameters must be supplied.
    deprecated: true
    mode: write
    args: provisioner_id_in text, worker_type_in text, expires_in timestamptz, last_date_active_in timestamptz, description_in text, stability_in text
    returns: table(provisioner_id text, worker_type text, expires timestamptz, last_date_active timestamptz, description text, stability text, etag uuid)
    body: |-
      begin
        return query update task_queues
        set
          expires = expires_in,
          last_date_active = last_date_active_in,
          description = description_in,
          stability = stability_in
        where
          task_queues.task_queue_id = provisioner_id_in || '/' || worker_type_in
        returning
          split_part(task_queues.task_queue_id, '/', 1) as provisioner_id,
          split_part(task_queues.task_queue_id, '/', 2) as worker_type,
          task_queues.expires,
          task_queues.last_date_active,
          task_queues.description,
          task_queues.stability,
          public.gen_random_uuid();
      end
  get_queue_worker_types:
    description: |-
      Get queue worker types ordered by `provisioner_id` and `worker_type`.
      If the pagination arguments are both NULL, all rows are returned.
      Otherwise, page_size rows are returned at offset page_offset.
    deprecated: true
    mode: read
    serviceName: queue
    args: provisioner_id_in text, worker_type_in text, expires_in timestamptz, page_size_in integer, page_offset_in integer
    returns: table(provisioner_id text, worker_type text, expires timestamptz, last_date_active timestamptz, description text, stability text, etag uuid)
    body: |-
      begin
        return query
        select
          split_part(task_queues.task_queue_id, '/', 1) as provisioner_id,
          split_part(task_queues.task_queue_id, '/', 2) as worker_type,
          task_queues.expires,
          task_queues.last_date_active,
          task_queues.description,
          task_queues.stability,
          public.gen_random_uuid()
        from task_queues
        where
          (split_part(task_queues.task_queue_id, '/', 1) = provisioner_id_in or provisioner_id_in is null) and
          (split_part(task_queues.task_queue_id, '/', 2) = worker_type_in or worker_type_in is null) and
          (task_queues.expires > expires_in or expires_in is null)
        order by task_queue_id
        limit get_page_limit(page_size_in)
        offset get_page_offset(page_offset_in);
      end
  expire_queue_worker_types:
    description: |-
      Expire queue worker types that come before `expires_in`.
      Returns a count of rows that have been deleted.
    deprecated: true
    mode: write
    serviceName: queue
    args: expires_in timestamptz
    returns: integer
    body: |-
      declare
        count integer;
      begin
        delete from task_queues
        where task_queues.expires < expires_in;
        if found then
          get diagnostics count = row_count;
          return count;
        end if;
        return 0;
      end
  create_queue_worker:
    description: |-
      Create a new queue worker.  Raises UNIQUE_VIOLATION if the worker already exists.
    deprecated: true
    mode: write
    serviceName: queue
    args: provisioner_id_in text, worker_type_in text, worker_group_in text, worker_id_in text, quarantine_until_in timestamptz, expires_in timestamptz, first_claim_in timestamptz, recent_tasks_in jsonb
    returns: uuid
    body: |-
      begin
        insert
          into queue_workers (task_queue_id, worker_group, worker_id, quarantine_until, expires, first_claim, recent_tasks)
          values (
            provisioner_id_in || '/' || worker_type_in,
            worker_group_in,
            worker_id_in,
            quarantine_until_in,
            expires_in,
            first_claim_in,
            recent_tasks_in
          );
          return public.gen_random_uuid();
      end
  get_queue_worker:
    description: |-
      Get a non-expired queue worker by provisioner_id, worker_type, worker_group, and worker_id.
      Workers are not considered expired until after their quarantine date expires.
    deprecated: true
    mode: read
    serviceName: queue
    args: provisioner_id_in text, worker_type_in text, worker_group_in text, worker_id_in text, expires_in timestamptz
    returns: table(provisioner_id text, worker_type text, worker_group text, worker_id text, quarantine_until timestamptz, expires timestamptz, first_claim timestamptz, recent_tasks jsonb, etag uuid)
    body: |-
      begin
        return query
        select
          split_part(queue_workers.task_queue_id, '/', 1) as provisioner_id,
          split_part(queue_workers.task_queue_id, '/', 2) as worker_type,
          queue_workers.worker_group,
          queue_workers.worker_id,
          queue_workers.quarantine_until,
          queue_workers.expires,
          queue_workers.first_claim,
          queue_workers.recent_tasks,
          public.gen_random_uuid()
        from queue_workers
        where
          queue_workers.task_queue_id = provisioner_id_in || '/' || worker_type_in and
          queue_workers.worker_group = worker_group_in and
          queue_workers.worker_id = worker_id_in and
          (queue_workers.expires > expires_in or queue_workers.quarantine_until > expires_in);
        end
  update_queue_worker:
    serviceName: queue
    description: |-
      Update a queue worker's quarantine_until, expires, and recent_tasks.
      All parameters must be supplied.
    deprecated: true
    mode: write
    args: provisioner_id_in text, worker_type_in text, worker_group_in text, worker_id_in text, quarantine_until_in timestamptz, expires_in timestamptz, recent_tasks_in jsonb
    returns: table(provisioner_id text, worker_type text, worker_group text, worker_id text, quarantine_until timestamptz, expires timestamptz, first_claim timestamptz, recent_tasks jsonb, etag uuid)
    body: |-
      begin
        return query update queue_workers
        set
          quarantine_until = quarantine_until_in,
          expires = expires_in,
          recent_tasks = recent_tasks_in
        where
          queue_workers.task_queue_id = provisioner_id_in || '/' || worker_type_in and
          queue_workers.worker_group = worker_group_in and
          queue_workers.worker_id = worker_id_in
        returning
          split_part(queue_workers.task_queue_id, '/', 1) as provisioner_id,
          split_part(queue_workers.task_queue_id, '/', 2) as worker_type,
          queue_workers.worker_group,
          queue_workers.worker_id,
          queue_workers.quarantine_until,
          queue_workers.expires,
          queue_workers.first_claim,
          queue_workers.recent_tasks,
          public.gen_random_uuid();
      end
  get_queue_workers:
    description: |-
      Get non-expired queue workers ordered by provisioner_id, worker_type, worker_group, and worker_id.
      Workers are not considered expired until after their quarantine date expires.
      If the pagination arguments are both NULL, all rows are returned.
      Otherwise, page_size rows are returned at offset page_offset.
    deprecated: true
    mode: read
    serviceName: queue
    args: provisioner_id_in text, worker_type_in text, expires_in timestamptz, page_size_in integer, page_offset_in integer
    returns: table(provisioner_id text, worker_type text, worker_group text, worker_id text, quarantine_until timestamptz, expires timestamptz, first_claim timestamptz, recent_tasks jsonb, etag uuid)
    body: |-
      begin
        return query
        select
          split_part(queue_workers.task_queue_id, '/', 1) as provisioner_id,
          split_part(queue_workers.task_queue_id, '/', 2) as worker_type,
          queue_workers.worker_group,
          queue_workers.worker_id,
          queue_workers.quarantine_until,
          queue_workers.expires,
          queue_workers.first_claim,
          queue_workers.recent_tasks,
          public.gen_random_uuid()
        from queue_workers
        where
          (split_part(queue_workers.task_queue_id, '/', 1) = provisioner_id_in or provisioner_id_in is null) and
          (split_part(queue_workers.task_queue_id, '/', 2) = worker_type_in or worker_type_in is null) and
          ((queue_workers.expires > expires_in and queue_workers.quarantine_until < expires_in) or get_queue_workers.expires_in is null)
        order by task_queue_id, worker_group, worker_id
        limit get_page_limit(page_size_in)
        offset get_page_offset(page_offset_in);
      end
  create_task_queue:
    description: |-
      Create a new task queue. Raises UNIQUE_VIOLATION if the task queue already exists.
    mode: write
    serviceName: queue
    args: task_queue_id_in text, expires_in timestamptz, last_date_active_in timestamptz, description_in text, stability_in text
    returns: uuid
    body: |-
      begin
        insert
          into task_queues (task_queue_id, expires, last_date_active, description, stability)
          values (
            task_queue_id_in,
            expires_in,
            last_date_active_in,
            description_in, stability_in
          );
          return public.gen_random_uuid();
      end
  get_task_queue:
    description: |-
      Get a non-expired task queue by task_queue_id.
    mode: read
    serviceName: queue
    args: task_queue_id_in text, expires_in timestamptz
    returns: table(task_queue_id text, expires timestamptz, last_date_active timestamptz, description text, stability text, etag uuid)
    body: |-
      begin
        return query
        select
          task_queues.task_queue_id,
          task_queues.expires,
          task_queues.last_date_active,
          task_queues.description,
          task_queues.stability,
          public.gen_random_uuid()
        from task_queues
        where
          task_queues.task_queue_id = task_queue_id_in and
          task_queues.expires > expires_in;
        end
  update_task_queue:
    serviceName: queue
    description: |-
      Update a task queue's expires, last_date_active, description, and stability.
      All parameters must be supplied.
    mode: write
    args: task_queue_id_in text, expires_in timestamptz, last_date_active_in timestamptz, description_in text, stability_in text
    returns: table(task_queue_id text, expires timestamptz, last_date_active timestamptz, description text, stability text, etag uuid)
    body: |-
      begin
        return query update task_queues
        set
          expires = expires_in,
          last_date_active = last_date_active_in,
          description = description_in,
          stability = stability_in
        where
          task_queues.task_queue_id = task_queue_id_in
        returning
          task_queues.task_queue_id,
          task_queues.expires,
          task_queues.last_date_active,
          task_queues.description,
          task_queues.stability,
          public.gen_random_uuid();
      end
  get_task_queues:
    description: |-
      Get task queues ordered by `task_queue_id`.
      If the pagination arguments are both NULL, all rows are returned.
      Otherwise, page_size rows are returned at offset page_offset.
    mode: read
    serviceName: queue
    args: task_queue_id_in text, expires_in timestamptz, page_size_in integer, page_offset_in integer
    returns: table(task_queue_id text, expires timestamptz, last_date_active timestamptz, description text, stability text, etag uuid)
    body: |-
      begin
        return query
        select
          task_queues.task_queue_id,
          task_queues.expires,
          task_queues.last_date_active,
          task_queues.description,
          task_queues.stability,
          public.gen_random_uuid()
        from task_queues
        where
          (task_queues.task_queue_id = task_queue_id_in or task_queue_id_in is null) and
          (task_queues.expires > expires_in or expires_in is null)
        order by task_queue_id
        limit get_page_limit(page_size_in)
        offset get_page_offset(page_offset_in);
      end
  expire_task_queues:
    description: |-
      Expire task queues that come before `expires_in`.
      Returns a count of rows that have been deleted.
    mode: write
    serviceName: queue
    args: expires_in timestamptz
    returns: integer
    body: |-
      declare
        count integer;
      begin
        delete from task_queues
        where task_queues.expires < expires_in;
        if found then
          get diagnostics count = row_count;
          return count;
        end if;
        return 0;
      end

  create_queue_worker_tqid:
    description: |-
      Create a new queue worker.  Raises UNIQUE_VIOLATION if the worker already exists.
    mode: write
    serviceName: queue
    args: task_queue_id_in text, worker_group_in text, worker_id_in text, quarantine_until_in timestamptz, expires_in timestamptz, first_claim_in timestamptz, recent_tasks_in jsonb
    returns: uuid
    body: |-
      begin
        insert
          into queue_workers (task_queue_id, worker_group, worker_id, quarantine_until, expires, first_claim, recent_tasks)
          values (
            task_queue_id_in,
            worker_group_in,
            worker_id_in,
            quarantine_until_in,
            expires_in,
            first_claim_in,
            recent_tasks_in
          );
          return public.gen_random_uuid();
      end
  get_queue_worker_tqid:
    description: |-
      Get a non-expired queue worker by task_queue_id, worker_group, and worker_id.
      Workers are not considered expired until after their quarantine date expires.
    mode: read
    serviceName: queue
    args: task_queue_id_in text, worker_group_in text, worker_id_in text, expires_in timestamptz
    returns: table(task_queue_id text, worker_group text, worker_id text, quarantine_until timestamptz, expires timestamptz, first_claim timestamptz, recent_tasks jsonb, etag uuid)
    body: |-
      begin
        return query
        select
          queue_workers.task_queue_id,
          queue_workers.worker_group,
          queue_workers.worker_id,
          queue_workers.quarantine_until,
          queue_workers.expires,
          queue_workers.first_claim,
          queue_workers.recent_tasks,
          public.gen_random_uuid()
        from queue_workers
        where
          queue_workers.task_queue_id = task_queue_id_in and
          queue_workers.worker_group = worker_group_in and
          queue_workers.worker_id = worker_id_in and
          (queue_workers.expires > expires_in or queue_workers.quarantine_until > expires_in);
        end
  update_queue_worker_tqid:
    serviceName: queue
    description: |-
      Update a queue worker's quarantine_until, expires, and recent_tasks.
      All parameters must be supplied.
    mode: write
    args: task_queue_id_in text, worker_group_in text, worker_id_in text, quarantine_until_in timestamptz, expires_in timestamptz, recent_tasks_in jsonb
    returns: table(task_queue_id text, worker_group text, worker_id text, quarantine_until timestamptz, expires timestamptz, first_claim timestamptz, recent_tasks jsonb, etag uuid)
    body: |-
      begin
        return query update queue_workers
        set
          quarantine_until = quarantine_until_in,
          expires = expires_in,
          recent_tasks = recent_tasks_in
        where
          queue_workers.task_queue_id = task_queue_id_in and
          queue_workers.worker_group = worker_group_in and
          queue_workers.worker_id = worker_id_in
        returning
          queue_workers.task_queue_id,
          queue_workers.worker_group,
          queue_workers.worker_id,
          queue_workers.quarantine_until,
          queue_workers.expires,
          queue_workers.first_claim,
          queue_workers.recent_tasks,
          public.gen_random_uuid();
      end
  get_queue_workers_tqid:
    description: |-
      Get non-expired queue workers ordered by task_queue_id, worker_group, and worker_id.
      Workers are not considered expired until after their quarantine date expires.
      If the pagination arguments are both NULL, all rows are returned.
      Otherwise, page_size rows are returned at offset page_offset.
    mode: read
    serviceName: queue
    args: task_queue_id_in text, expires_in timestamptz, page_size_in integer, page_offset_in integer
    returns: table(task_queue_id text, worker_group text, worker_id text, quarantine_until timestamptz, expires timestamptz, first_claim timestamptz, recent_tasks jsonb, etag uuid)
    body: |-
      begin
        return query
        select
          queue_workers.task_queue_id,
          queue_workers.worker_group,
          queue_workers.worker_id,
          queue_workers.quarantine_until,
          queue_workers.expires,
          queue_workers.first_claim,
          queue_workers.recent_tasks,
          public.gen_random_uuid()
        from queue_workers
        where
          (queue_workers.task_queue_id = task_queue_id_in or get_queue_workers_tqid.task_queue_id_in is null) and
          ((queue_workers.expires > expires_in and queue_workers.quarantine_until < expires_in) or get_queue_workers_tqid.expires_in is null)
        order by task_queue_id, worker_group, worker_id
        limit get_page_limit(page_size_in)
        offset get_page_offset(page_offset_in);
      end
  create_queue_provisioner:
    description: |-
      Create a new queue provisioner.  Raises UNIQUE_VIOLATION if the provisioner already exists.
    deprecated: true
    mode: write
    serviceName: queue
    args: provisioner_id_in text, expires_in timestamptz, last_date_active_in timestamptz, description_in text, stability_in text, actions_in jsonb
    returns: uuid
    body: |-
      begin
        return public.gen_random_uuid();
      end
  get_queue_provisioner:
    description: |-
      Get a queue provisioner by provisioner_id.
    deprecated: true
    mode: read
    serviceName: queue
    args: provisioner_id_in text, expires_in timestamptz
    returns: table(provisioner_id text, expires timestamptz, last_date_active timestamptz, description text, stability text, actions jsonb, etag uuid)
    body: |-
      begin
        return query
        select
          split_part(task_queue_id, '/', 1) as provisioner_id,
          max(task_queues.expires),
          max(task_queues.last_date_active),
          '' as description,
          'experimental' as stability,
          '[]'::jsonb as actions,
          public.gen_random_uuid()
        from task_queues
        group by provisioner_id
        having
          split_part(task_queue_id, '/', 1) = provisioner_id_in and
          max(task_queues.expires) > expires_in;
      end
  update_queue_provisioner:
    serviceName: queue
    description: |-
      Update a queue provisioner's expires, last_date_active, description, stability, and actions.
      All parameters must be supplied.
    deprecated: true
    mode: write
    args: provisioner_id_in text, expires_in timestamptz, last_date_active_in timestamptz, description_in text, stability_in text, actions_in jsonb
    returns: table(provisioner_id text, expires timestamptz, last_date_active timestamptz, description text, stability text, actions jsonb, etag uuid)
    body: |-
      begin
        return query
        select
          split_part(task_queue_id, '/', 1) as provisioner_id,
          max(task_queues.expires),
          max(task_queues.last_date_active),
          '' as description,
          'experimental' as stability,
          '[]'::jsonb as actions,
          public.gen_random_uuid()
        from task_queues
        group by provisioner_id
        having
          split_part(task_queue_id, '/', 1) = provisioner_id_in;
      end
  get_queue_provisioners:
    description: |-
      Get queue provisioners ordered by `provisioner_id`.
      If the pagination arguments are both NULL, all rows are returned.
      Otherwise, page_size rows are returned at offset page_offset.
    deprecated: true
    mode: read
    serviceName: queue
    args: expires_in timestamptz, page_size_in integer, page_offset_in integer
    returns: table(provisioner_id text, expires timestamptz, last_date_active timestamptz, description text, stability text, actions jsonb, etag uuid)
    body: |-
      begin
        return query
        select
          split_part(task_queues.task_queue_id, '/', 1) as provisioner_id,
          max(task_queues.expires),
          max(task_queues.last_date_active),
          '' as description,
          'experimental' as stability,
          '[]'::jsonb as actions,
          public.gen_random_uuid()
        from task_queues
        group by provisioner_id
        having (max(task_queues.expires) > expires_in or expires_in is null)
        order by provisioner_id
        limit get_page_limit(page_size_in)
        offset get_page_offset(page_offset_in);
      end
  expire_queue_provisioners:
    description: |-
      Expire provisioners that come before `expires_in`.
      Returns a count of rows that have been deleted.
    deprecated: true
    mode: write
    serviceName: queue
    args: expires_in timestamptz
    returns: integer
    body: |-
      begin
        return 0;
      end