-module(m25@internal@sql). -compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch]). -define(FILEPATH, "src/m25/internal/sql.gleam"). -export([finalise_job_reservations/3, heartbeat/4, retry_if_needed/2, succeed_job/3, cancel_job/2, cleanup_stuck_reservations/3, error_job/3, fail_job/3, get_job/2, insert_job/12, reserve_jobs/3, time_out_jobs/2]). -export_type([cancel_job_row/0, cleanup_stuck_reservations_row/0, error_job_row/0, fail_job_row/0, finalise_job_reservations_row/0, get_job_row/0, heartbeat_row/0, insert_job_row/0, reserve_jobs_row/0, time_out_jobs_row/0]). -if(?OTP_RELEASE >= 27). -define(MODULEDOC(Str), -moduledoc(Str)). -define(DOC(Str), -doc(Str)). -else. -define(MODULEDOC(Str), -compile([])). -define(DOC(Str), -compile([])). -endif. ?MODULEDOC(false). -type cancel_job_row() :: {cancel_job_row, youid@uuid:uuid(), binary(), gleam@time@timestamp:timestamp(), gleam@time@timestamp:timestamp(), binary(), gleam@time@timestamp:timestamp(), gleam@time@timestamp:timestamp(), gleam@time@timestamp:timestamp(), gleam@time@timestamp:timestamp(), integer(), binary(), binary(), gleam@time@timestamp:timestamp(), gleam@time@timestamp:timestamp(), binary(), binary(), integer(), integer(), youid@uuid:uuid(), youid@uuid:uuid(), integer(), binary(), binary()}. -type cleanup_stuck_reservations_row() :: {cleanup_stuck_reservations_row, youid@uuid:uuid(), binary()}. -type error_job_row() :: {error_job_row, youid@uuid:uuid(), binary(), binary(), binary(), integer(), integer(), gleam@option:option(youid@uuid:uuid()), gleam@option:option(youid@uuid:uuid()), integer(), gleam@option:option(binary())}. -type fail_job_row() :: {fail_job_row, youid@uuid:uuid(), binary(), binary(), binary(), integer(), integer(), gleam@option:option(youid@uuid:uuid()), gleam@option:option(youid@uuid:uuid()), integer(), gleam@option:option(binary())}. -type finalise_job_reservations_row() :: {finalise_job_reservations_row, integer(), integer()}. -type get_job_row() :: {get_job_row, youid@uuid:uuid(), binary(), gleam@time@timestamp:timestamp(), gleam@time@timestamp:timestamp(), binary(), gleam@time@timestamp:timestamp(), gleam@time@timestamp:timestamp(), gleam@time@timestamp:timestamp(), gleam@time@timestamp:timestamp(), integer(), binary(), gleam@option:option(binary()), gleam@time@timestamp:timestamp(), gleam@time@timestamp:timestamp(), gleam@option:option(binary()), gleam@option:option(binary()), integer(), integer(), gleam@option:option(youid@uuid:uuid()), gleam@option:option(youid@uuid:uuid()), integer(), gleam@option:option(binary())}. -type heartbeat_row() :: {heartbeat_row, boolean(), boolean()}. -type insert_job_row() :: {insert_job_row, youid@uuid:uuid(), binary(), gleam@time@timestamp:timestamp(), gleam@time@timestamp:timestamp(), binary(), gleam@time@timestamp:timestamp(), gleam@time@timestamp:timestamp(), gleam@time@timestamp:timestamp(), gleam@time@timestamp:timestamp(), integer(), binary(), gleam@option:option(binary()), gleam@time@timestamp:timestamp(), gleam@time@timestamp:timestamp(), gleam@option:option(binary()), gleam@option:option(binary()), integer(), integer(), gleam@option:option(youid@uuid:uuid()), gleam@option:option(youid@uuid:uuid()), integer(), gleam@option:option(binary())}. -type reserve_jobs_row() :: {reserve_jobs_row, youid@uuid:uuid(), binary(), gleam@time@timestamp:timestamp(), gleam@time@timestamp:timestamp(), binary(), gleam@time@timestamp:timestamp(), gleam@time@timestamp:timestamp(), gleam@time@timestamp:timestamp(), gleam@time@timestamp:timestamp(), integer(), binary(), gleam@option:option(binary()), gleam@time@timestamp:timestamp(), gleam@time@timestamp:timestamp(), gleam@option:option(binary()), gleam@option:option(binary()), integer(), integer(), gleam@option:option(youid@uuid:uuid()), gleam@option:option(youid@uuid:uuid()), integer(), gleam@option:option(binary())}. -type time_out_jobs_row() :: {time_out_jobs_row, youid@uuid:uuid(), binary(), binary(), integer(), integer(), gleam@option:option(youid@uuid:uuid()), gleam@option:option(youid@uuid:uuid()), integer(), gleam@option:option(binary())}. -file("src/m25/internal/sql.gleam", 405). ?DOC(false). -spec finalise_job_reservations( pog:connection(), list(youid@uuid:uuid()), list(youid@uuid:uuid()) ) -> {ok, pog:returned(finalise_job_reservations_row())} | {error, pog:query_error()}. finalise_job_reservations(Db, Arg_1, Arg_2) -> Decoder = begin gleam@dynamic@decode:field( 0, {decoder, fun gleam@dynamic@decode:decode_int/1}, fun(Successful_count) -> gleam@dynamic@decode:field( 1, {decoder, fun gleam@dynamic@decode:decode_int/1}, fun(Failed_count) -> gleam@dynamic@decode:success( {finalise_job_reservations_row, Successful_count, Failed_count} ) end ) end ) end, _pipe = <<"-- Promote successful reservations to executing, revert failures to pending with successful_jobs as ( update m25.job set started_at = now(), deadline = now() + timeout where id = any($1) and status = 'reserved' returning id ), failed_jobs as ( update m25.job set reserved_at = null where id = any($2) and status = 'reserved' returning id ) select (select count(*) from successful_jobs) as successful_count, (select count(*) from failed_jobs) as failed_count; "/utf8>>, _pipe@1 = pog:'query'(_pipe), _pipe@2 = pog:parameter( _pipe@1, pog:array( fun(Value) -> pog_ffi:coerce(youid@uuid:to_string(Value)) end, Arg_1 ) ), _pipe@3 = pog:parameter( _pipe@2, pog:array( fun(Value@1) -> pog_ffi:coerce(youid@uuid:to_string(Value@1)) end, Arg_2 ) ), _pipe@4 = pog:returning(_pipe@3, Decoder), pog:execute(_pipe@4, Db). -file("src/m25/internal/sql.gleam", 583). ?DOC(false). -spec heartbeat(pog:connection(), youid@uuid:uuid(), integer(), float()) -> {ok, pog:returned(heartbeat_row())} | {error, pog:query_error()}. heartbeat(Db, Arg_1, Arg_2, Arg_3) -> Decoder = begin gleam@dynamic@decode:field( 0, {decoder, fun gleam@dynamic@decode:decode_bool/1}, fun(Heartbeat_timed_out) -> gleam@dynamic@decode:field( 1, {decoder, fun gleam@dynamic@decode:decode_bool/1}, fun(Deadline_passed) -> gleam@dynamic@decode:success( {heartbeat_row, Heartbeat_timed_out, Deadline_passed} ) end ) end ) end, _pipe = <<"update m25.job set latest_heartbeat_at = now() where id = $1 returning ( -- This is the only way to access the old value of the row select now() - coalesce(latest_heartbeat_at, now()) > $2::int * make_interval(secs => $3) as heartbeat_timed_out from m25.job where id = $1 ) as heartbeat_timed_out, now() > deadline as deadline_passed; "/utf8>>, _pipe@1 = pog:'query'(_pipe), _pipe@2 = pog:parameter( _pipe@1, pog_ffi:coerce(youid@uuid:to_string(Arg_1)) ), _pipe@3 = pog:parameter(_pipe@2, pog_ffi:coerce(Arg_2)), _pipe@4 = pog:parameter(_pipe@3, pog_ffi:coerce(Arg_3)), _pipe@5 = pog:returning(_pipe@4, Decoder), pog:execute(_pipe@5, Db). -file("src/m25/internal/sql.gleam", 917). ?DOC(false). -spec retry_if_needed(pog:connection(), list(youid@uuid:uuid())) -> {ok, pog:returned(nil)} | {error, pog:query_error()}. retry_if_needed(Db, Arg_1) -> Decoder = gleam@dynamic@decode:map( {decoder, fun gleam@dynamic@decode:decode_dynamic/1}, fun(_) -> nil end ), _pipe = <<"insert into m25.job ( queue_name, scheduled_at, input, timeout, attempt, max_attempts, original_attempt_id, previous_attempt_id, retry_delay, unique_key ) ( select queue_name, now() + retry_delay as scheduled_at, input, timeout, attempt + 1 as attempt, max_attempts, coalesce(original_attempt_id, id) as original_attempt_id, id as previous_attempt_id, retry_delay, unique_key from m25.job where id = any($1) and attempt < max_attempts ); "/utf8>>, _pipe@1 = pog:'query'(_pipe), _pipe@2 = pog:parameter( _pipe@1, pog:array( fun(Value) -> pog_ffi:coerce(youid@uuid:to_string(Value)) end, Arg_1 ) ), _pipe@3 = pog:returning(_pipe@2, Decoder), pog:execute(_pipe@3, Db). -file("src/m25/internal/sql.gleam", 963). ?DOC(false). -spec succeed_job(pog:connection(), youid@uuid:uuid(), gleam@json:json()) -> {ok, pog:returned(nil)} | {error, pog:query_error()}. succeed_job(Db, Arg_1, Arg_2) -> Decoder = gleam@dynamic@decode:map( {decoder, fun gleam@dynamic@decode:decode_dynamic/1}, fun(_) -> nil end ), _pipe = <<"update m25.job set output = $2, finished_at = now() where id = $1; "/utf8>>, _pipe@1 = pog:'query'(_pipe), _pipe@2 = pog:parameter( _pipe@1, pog_ffi:coerce(youid@uuid:to_string(Arg_1)) ), _pipe@3 = pog:parameter( _pipe@2, pog_ffi:coerce(gleam@json:to_string(Arg_2)) ), _pipe@4 = pog:returning(_pipe@3, Decoder), pog:execute(_pipe@4, Db). -file("src/m25/internal/sql.gleam", 1058). ?DOC(false). -spec uuid_decoder() -> gleam@dynamic@decode:decoder(youid@uuid:uuid()). uuid_decoder() -> gleam@dynamic@decode:then( {decoder, fun gleam@dynamic@decode:decode_bit_array/1}, fun(Bit_array) -> case youid@uuid:from_bit_array(Bit_array) of {ok, Uuid} -> gleam@dynamic@decode:success(Uuid); {error, _} -> gleam@dynamic@decode:failure( youid@uuid:v7(), <<"Uuid"/utf8>> ) end end ). -file("src/m25/internal/sql.gleam", 53). ?DOC(false). -spec cancel_job(pog:connection(), youid@uuid:uuid()) -> {ok, pog:returned(cancel_job_row())} | {error, pog:query_error()}. cancel_job(Db, Arg_1) -> Decoder = begin gleam@dynamic@decode:field( 0, uuid_decoder(), fun(Id) -> gleam@dynamic@decode:field( 1, {decoder, fun gleam@dynamic@decode:decode_string/1}, fun(Queue_name) -> gleam@dynamic@decode:field( 2, pog:timestamp_decoder(), fun(Created_at) -> gleam@dynamic@decode:field( 3, pog:timestamp_decoder(), fun(Scheduled_at) -> gleam@dynamic@decode:field( 4, {decoder, fun gleam@dynamic@decode:decode_string/1}, fun(Input) -> gleam@dynamic@decode:field( 5, pog:timestamp_decoder(), fun(Reserved_at) -> gleam@dynamic@decode:field( 6, pog:timestamp_decoder( ), fun(Started_at) -> gleam@dynamic@decode:field( 7, pog:timestamp_decoder( ), fun( Cancelled_at ) -> gleam@dynamic@decode:field( 8, pog:timestamp_decoder( ), fun( Finished_at ) -> gleam@dynamic@decode:field( 9, {decoder, fun gleam@dynamic@decode:decode_int/1}, fun( Timeout ) -> gleam@dynamic@decode:field( 10, {decoder, fun gleam@dynamic@decode:decode_string/1}, fun( Status ) -> gleam@dynamic@decode:field( 11, {decoder, fun gleam@dynamic@decode:decode_string/1}, fun( Output ) -> gleam@dynamic@decode:field( 12, pog:timestamp_decoder( ), fun( Deadline ) -> gleam@dynamic@decode:field( 13, pog:timestamp_decoder( ), fun( Latest_heartbeat_at ) -> gleam@dynamic@decode:field( 14, {decoder, fun gleam@dynamic@decode:decode_string/1}, fun( Failure_reason ) -> gleam@dynamic@decode:field( 15, {decoder, fun gleam@dynamic@decode:decode_string/1}, fun( Error_data ) -> gleam@dynamic@decode:field( 16, {decoder, fun gleam@dynamic@decode:decode_int/1}, fun( Attempt ) -> gleam@dynamic@decode:field( 17, {decoder, fun gleam@dynamic@decode:decode_int/1}, fun( Max_attempts ) -> gleam@dynamic@decode:field( 18, uuid_decoder( ), fun( Original_attempt_id ) -> gleam@dynamic@decode:field( 19, uuid_decoder( ), fun( Previous_attempt_id ) -> gleam@dynamic@decode:field( 20, {decoder, fun gleam@dynamic@decode:decode_int/1}, fun( Retry_delay ) -> gleam@dynamic@decode:field( 21, {decoder, fun gleam@dynamic@decode:decode_string/1}, fun( Unique_key ) -> gleam@dynamic@decode:field( 22, {decoder, fun gleam@dynamic@decode:decode_string/1}, fun( Cancel_outcome ) -> gleam@dynamic@decode:success( {cancel_job_row, Id, Queue_name, Created_at, Scheduled_at, Input, Reserved_at, Started_at, Cancelled_at, Finished_at, Timeout, Status, Output, Deadline, Latest_heartbeat_at, Failure_reason, Error_data, Attempt, Max_attempts, Original_attempt_id, Previous_attempt_id, Retry_delay, Unique_key, Cancel_outcome} ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end, _pipe = <<"-- Cancel a job only if it's pending, and distinguish outcomes in one query with target as ( select id, queue_name, created_at::timestamp, scheduled_at::timestamp, input, reserved_at::timestamp, started_at::timestamp, cancelled_at::timestamp, finished_at::timestamp, extract(epoch from timeout)::int as timeout, status, output, deadline::timestamp, latest_heartbeat_at::timestamp, failure_reason, error_data, attempt, max_attempts, original_attempt_id, previous_attempt_id, extract(epoch from retry_delay)::int as retry_delay, unique_key from m25.job where id = $1 ), updated as ( update m25.job j set cancelled_at = now() from target t where j.id = t.id and t.status = 'pending' returning j.id, j.queue_name, j.created_at::timestamp, j.scheduled_at::timestamp, j.input, j.reserved_at::timestamp, j.started_at::timestamp, j.cancelled_at::timestamp, j.finished_at::timestamp, extract(epoch from j.timeout)::int as timeout, j.status, j.output, j.deadline::timestamp, j.latest_heartbeat_at::timestamp, j.failure_reason, j.error_data, j.attempt, j.max_attempts, j.original_attempt_id, j.previous_attempt_id, extract(epoch from j.retry_delay)::int as retry_delay, j.unique_key ) select coalesce(u.id, t.id) as id, coalesce(u.queue_name, t.queue_name) as queue_name, coalesce(u.created_at, t.created_at)::timestamp as created_at, coalesce(u.scheduled_at, t.scheduled_at)::timestamp as scheduled_at, coalesce(u.input, t.input) as input, coalesce(u.reserved_at, t.reserved_at)::timestamp as reserved_at, coalesce(u.started_at, t.started_at)::timestamp as started_at, coalesce(u.cancelled_at, t.cancelled_at)::timestamp as cancelled_at, coalesce(u.finished_at, t.finished_at)::timestamp as finished_at, coalesce(u.timeout, t.timeout)::int as timeout, coalesce(u.status, t.status) as status, coalesce(u.output, t.output) as output, coalesce(u.deadline, t.deadline)::timestamp as deadline, coalesce(u.latest_heartbeat_at, t.latest_heartbeat_at)::timestamp as latest_heartbeat_at, coalesce(u.failure_reason, t.failure_reason) as failure_reason, coalesce(u.error_data, t.error_data) as error_data, coalesce(u.attempt, t.attempt) as attempt, coalesce(u.max_attempts, t.max_attempts) as max_attempts, coalesce(u.original_attempt_id, t.original_attempt_id) as original_attempt_id, coalesce(u.previous_attempt_id, t.previous_attempt_id) as previous_attempt_id, coalesce(u.retry_delay, t.retry_delay) as retry_delay, coalesce(u.unique_key, t.unique_key) as unique_key, case when u.id is not null then 'cancelled' else 'not_pending' end as cancel_outcome from target t left join updated u on u.id = t.id; "/utf8>>, _pipe@1 = pog:'query'(_pipe), _pipe@2 = pog:parameter( _pipe@1, pog_ffi:coerce(youid@uuid:to_string(Arg_1)) ), _pipe@3 = pog:returning(_pipe@2, Decoder), pog:execute(_pipe@3, Db). -file("src/m25/internal/sql.gleam", 211). ?DOC(false). -spec cleanup_stuck_reservations(pog:connection(), binary(), float()) -> {ok, pog:returned(cleanup_stuck_reservations_row())} | {error, pog:query_error()}. cleanup_stuck_reservations(Db, Arg_1, Arg_2) -> Decoder = begin gleam@dynamic@decode:field( 0, uuid_decoder(), fun(Id) -> gleam@dynamic@decode:field( 1, {decoder, fun gleam@dynamic@decode:decode_string/1}, fun(Queue_name) -> gleam@dynamic@decode:success( {cleanup_stuck_reservations_row, Id, Queue_name} ) end ) end ) end, _pipe = <<"update m25.job set reserved_at = null where status = 'reserved' and queue_name = $1 and reserved_at < now() - make_interval(secs => $2) returning id, queue_name; "/utf8>>, _pipe@1 = pog:'query'(_pipe), _pipe@2 = pog:parameter(_pipe@1, pog_ffi:coerce(Arg_1)), _pipe@3 = pog:parameter(_pipe@2, pog_ffi:coerce(Arg_2)), _pipe@4 = pog:returning(_pipe@3, Decoder), pog:execute(_pipe@4, Db). -file("src/m25/internal/sql.gleam", 260). ?DOC(false). -spec error_job(pog:connection(), youid@uuid:uuid(), gleam@json:json()) -> {ok, pog:returned(error_job_row())} | {error, pog:query_error()}. error_job(Db, Arg_1, Arg_2) -> Decoder = begin gleam@dynamic@decode:field( 0, uuid_decoder(), fun(Id) -> gleam@dynamic@decode:field( 1, {decoder, fun gleam@dynamic@decode:decode_string/1}, fun(Queue_name) -> gleam@dynamic@decode:field( 2, {decoder, fun gleam@dynamic@decode:decode_string/1}, fun(Status) -> gleam@dynamic@decode:field( 3, {decoder, fun gleam@dynamic@decode:decode_string/1}, fun(Input) -> gleam@dynamic@decode:field( 4, {decoder, fun gleam@dynamic@decode:decode_int/1}, fun(Attempt) -> gleam@dynamic@decode:field( 5, {decoder, fun gleam@dynamic@decode:decode_int/1}, fun(Max_attempts) -> gleam@dynamic@decode:field( 6, gleam@dynamic@decode:optional( uuid_decoder() ), fun( Original_attempt_id ) -> gleam@dynamic@decode:field( 7, gleam@dynamic@decode:optional( uuid_decoder( ) ), fun( Previous_attempt_id ) -> gleam@dynamic@decode:field( 8, {decoder, fun gleam@dynamic@decode:decode_int/1}, fun( Retry_delay ) -> gleam@dynamic@decode:field( 9, gleam@dynamic@decode:optional( {decoder, fun gleam@dynamic@decode:decode_string/1} ), fun( Unique_key ) -> gleam@dynamic@decode:success( {error_job_row, Id, Queue_name, Status, Input, Attempt, Max_attempts, Original_attempt_id, Previous_attempt_id, Retry_delay, Unique_key} ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end, _pipe = <<"update m25.job set failure_reason = 'error', error_data = $2, finished_at = now() where id = $1 returning id, queue_name, status, input, attempt, max_attempts, original_attempt_id, previous_attempt_id, -- TODO: use duration once supported in Squirrel extract(epoch from retry_delay)::int as retry_delay, unique_key; "/utf8>>, _pipe@1 = pog:'query'(_pipe), _pipe@2 = pog:parameter( _pipe@1, pog_ffi:coerce(youid@uuid:to_string(Arg_1)) ), _pipe@3 = pog:parameter( _pipe@2, pog_ffi:coerce(gleam@json:to_string(Arg_2)) ), _pipe@4 = pog:returning(_pipe@3, Decoder), pog:execute(_pipe@4, Db). -file("src/m25/internal/sql.gleam", 339). ?DOC(false). -spec fail_job(pog:connection(), youid@uuid:uuid(), binary()) -> {ok, pog:returned(fail_job_row())} | {error, pog:query_error()}. fail_job(Db, Arg_1, Arg_2) -> Decoder = begin gleam@dynamic@decode:field( 0, uuid_decoder(), fun(Id) -> gleam@dynamic@decode:field( 1, {decoder, fun gleam@dynamic@decode:decode_string/1}, fun(Queue_name) -> gleam@dynamic@decode:field( 2, {decoder, fun gleam@dynamic@decode:decode_string/1}, fun(Status) -> gleam@dynamic@decode:field( 3, {decoder, fun gleam@dynamic@decode:decode_string/1}, fun(Input) -> gleam@dynamic@decode:field( 4, {decoder, fun gleam@dynamic@decode:decode_int/1}, fun(Attempt) -> gleam@dynamic@decode:field( 5, {decoder, fun gleam@dynamic@decode:decode_int/1}, fun(Max_attempts) -> gleam@dynamic@decode:field( 6, gleam@dynamic@decode:optional( uuid_decoder() ), fun( Original_attempt_id ) -> gleam@dynamic@decode:field( 7, gleam@dynamic@decode:optional( uuid_decoder( ) ), fun( Previous_attempt_id ) -> gleam@dynamic@decode:field( 8, {decoder, fun gleam@dynamic@decode:decode_int/1}, fun( Retry_delay ) -> gleam@dynamic@decode:field( 9, gleam@dynamic@decode:optional( {decoder, fun gleam@dynamic@decode:decode_string/1} ), fun( Unique_key ) -> gleam@dynamic@decode:success( {fail_job_row, Id, Queue_name, Status, Input, Attempt, Max_attempts, Original_attempt_id, Previous_attempt_id, Retry_delay, Unique_key} ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end, _pipe = <<"update m25.job set failure_reason = $2, finished_at = now() where id = $1 returning id, queue_name, status, input, attempt, max_attempts, original_attempt_id, previous_attempt_id, -- TODO: use duration once supported in Squirrel extract(epoch from retry_delay)::int as retry_delay, unique_key; "/utf8>>, _pipe@1 = pog:'query'(_pipe), _pipe@2 = pog:parameter( _pipe@1, pog_ffi:coerce(youid@uuid:to_string(Arg_1)) ), _pipe@3 = pog:parameter(_pipe@2, pog_ffi:coerce(Arg_2)), _pipe@4 = pog:returning(_pipe@3, Decoder), pog:execute(_pipe@4, Db). -file("src/m25/internal/sql.gleam", 486). ?DOC(false). -spec get_job(pog:connection(), youid@uuid:uuid()) -> {ok, pog:returned(get_job_row())} | {error, pog:query_error()}. get_job(Db, Arg_1) -> Decoder = begin gleam@dynamic@decode:field( 0, uuid_decoder(), fun(Id) -> gleam@dynamic@decode:field( 1, {decoder, fun gleam@dynamic@decode:decode_string/1}, fun(Queue_name) -> gleam@dynamic@decode:field( 2, pog:timestamp_decoder(), fun(Created_at) -> gleam@dynamic@decode:field( 3, pog:timestamp_decoder(), fun(Scheduled_at) -> gleam@dynamic@decode:field( 4, {decoder, fun gleam@dynamic@decode:decode_string/1}, fun(Input) -> gleam@dynamic@decode:field( 5, pog:timestamp_decoder(), fun(Reserved_at) -> gleam@dynamic@decode:field( 6, pog:timestamp_decoder( ), fun(Started_at) -> gleam@dynamic@decode:field( 7, pog:timestamp_decoder( ), fun( Cancelled_at ) -> gleam@dynamic@decode:field( 8, pog:timestamp_decoder( ), fun( Finished_at ) -> gleam@dynamic@decode:field( 9, {decoder, fun gleam@dynamic@decode:decode_int/1}, fun( Timeout ) -> gleam@dynamic@decode:field( 10, {decoder, fun gleam@dynamic@decode:decode_string/1}, fun( Status ) -> gleam@dynamic@decode:field( 11, gleam@dynamic@decode:optional( {decoder, fun gleam@dynamic@decode:decode_string/1} ), fun( Output ) -> gleam@dynamic@decode:field( 12, pog:timestamp_decoder( ), fun( Deadline ) -> gleam@dynamic@decode:field( 13, pog:timestamp_decoder( ), fun( Latest_heartbeat_at ) -> gleam@dynamic@decode:field( 14, gleam@dynamic@decode:optional( {decoder, fun gleam@dynamic@decode:decode_string/1} ), fun( Failure_reason ) -> gleam@dynamic@decode:field( 15, gleam@dynamic@decode:optional( {decoder, fun gleam@dynamic@decode:decode_string/1} ), fun( Error_data ) -> gleam@dynamic@decode:field( 16, {decoder, fun gleam@dynamic@decode:decode_int/1}, fun( Attempt ) -> gleam@dynamic@decode:field( 17, {decoder, fun gleam@dynamic@decode:decode_int/1}, fun( Max_attempts ) -> gleam@dynamic@decode:field( 18, gleam@dynamic@decode:optional( uuid_decoder( ) ), fun( Original_attempt_id ) -> gleam@dynamic@decode:field( 19, gleam@dynamic@decode:optional( uuid_decoder( ) ), fun( Previous_attempt_id ) -> gleam@dynamic@decode:field( 20, {decoder, fun gleam@dynamic@decode:decode_int/1}, fun( Retry_delay ) -> gleam@dynamic@decode:field( 21, gleam@dynamic@decode:optional( {decoder, fun gleam@dynamic@decode:decode_string/1} ), fun( Unique_key ) -> gleam@dynamic@decode:success( {get_job_row, Id, Queue_name, Created_at, Scheduled_at, Input, Reserved_at, Started_at, Cancelled_at, Finished_at, Timeout, Status, Output, Deadline, Latest_heartbeat_at, Failure_reason, Error_data, Attempt, Max_attempts, Original_attempt_id, Previous_attempt_id, Retry_delay, Unique_key} ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end, _pipe = <<"select id, queue_name, created_at::timestamp, scheduled_at::timestamp, input, reserved_at::timestamp, started_at::timestamp, cancelled_at::timestamp, finished_at::timestamp, extract(epoch from timeout)::int as timeout, status, output, deadline::timestamp, latest_heartbeat_at::timestamp, failure_reason, error_data, attempt, max_attempts, original_attempt_id, previous_attempt_id, extract(epoch from retry_delay)::int as retry_delay, unique_key from m25.job where id = $1; "/utf8>>, _pipe@1 = pog:'query'(_pipe), _pipe@2 = pog:parameter( _pipe@1, pog_ffi:coerce(youid@uuid:to_string(Arg_1)) ), _pipe@3 = pog:returning(_pipe@2, Decoder), pog:execute(_pipe@3, Db). -file("src/m25/internal/sql.gleam", 651). ?DOC(false). -spec insert_job( pog:connection(), youid@uuid:uuid(), binary(), float(), gleam@json:json(), float(), integer(), integer(), youid@uuid:uuid(), youid@uuid:uuid(), float(), binary() ) -> {ok, pog:returned(insert_job_row())} | {error, pog:query_error()}. insert_job( Db, Arg_1, Arg_2, Arg_3, Arg_4, Arg_5, Arg_6, Arg_7, Arg_8, Arg_9, Arg_10, Arg_11 ) -> Decoder = begin gleam@dynamic@decode:field( 0, uuid_decoder(), fun(Id) -> gleam@dynamic@decode:field( 1, {decoder, fun gleam@dynamic@decode:decode_string/1}, fun(Queue_name) -> gleam@dynamic@decode:field( 2, pog:timestamp_decoder(), fun(Created_at) -> gleam@dynamic@decode:field( 3, pog:timestamp_decoder(), fun(Scheduled_at) -> gleam@dynamic@decode:field( 4, {decoder, fun gleam@dynamic@decode:decode_string/1}, fun(Input) -> gleam@dynamic@decode:field( 5, pog:timestamp_decoder(), fun(Reserved_at) -> gleam@dynamic@decode:field( 6, pog:timestamp_decoder( ), fun(Started_at) -> gleam@dynamic@decode:field( 7, pog:timestamp_decoder( ), fun( Cancelled_at ) -> gleam@dynamic@decode:field( 8, pog:timestamp_decoder( ), fun( Finished_at ) -> gleam@dynamic@decode:field( 9, {decoder, fun gleam@dynamic@decode:decode_int/1}, fun( Timeout ) -> gleam@dynamic@decode:field( 10, {decoder, fun gleam@dynamic@decode:decode_string/1}, fun( Status ) -> gleam@dynamic@decode:field( 11, gleam@dynamic@decode:optional( {decoder, fun gleam@dynamic@decode:decode_string/1} ), fun( Output ) -> gleam@dynamic@decode:field( 12, pog:timestamp_decoder( ), fun( Deadline ) -> gleam@dynamic@decode:field( 13, pog:timestamp_decoder( ), fun( Latest_heartbeat_at ) -> gleam@dynamic@decode:field( 14, gleam@dynamic@decode:optional( {decoder, fun gleam@dynamic@decode:decode_string/1} ), fun( Failure_reason ) -> gleam@dynamic@decode:field( 15, gleam@dynamic@decode:optional( {decoder, fun gleam@dynamic@decode:decode_string/1} ), fun( Error_data ) -> gleam@dynamic@decode:field( 16, {decoder, fun gleam@dynamic@decode:decode_int/1}, fun( Attempt ) -> gleam@dynamic@decode:field( 17, {decoder, fun gleam@dynamic@decode:decode_int/1}, fun( Max_attempts ) -> gleam@dynamic@decode:field( 18, gleam@dynamic@decode:optional( uuid_decoder( ) ), fun( Original_attempt_id ) -> gleam@dynamic@decode:field( 19, gleam@dynamic@decode:optional( uuid_decoder( ) ), fun( Previous_attempt_id ) -> gleam@dynamic@decode:field( 20, {decoder, fun gleam@dynamic@decode:decode_int/1}, fun( Retry_delay ) -> gleam@dynamic@decode:field( 21, gleam@dynamic@decode:optional( {decoder, fun gleam@dynamic@decode:decode_string/1} ), fun( Unique_key ) -> gleam@dynamic@decode:success( {insert_job_row, Id, Queue_name, Created_at, Scheduled_at, Input, Reserved_at, Started_at, Cancelled_at, Finished_at, Timeout, Status, Output, Deadline, Latest_heartbeat_at, Failure_reason, Error_data, Attempt, Max_attempts, Original_attempt_id, Previous_attempt_id, Retry_delay, Unique_key} ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end, _pipe = <<"insert into m25.job ( id, queue_name, scheduled_at, input, timeout, attempt, max_attempts, original_attempt_id, previous_attempt_id, retry_delay, unique_key ) values ( $1, $2, to_timestamp($3), $4, make_interval(secs => $5), $6, $7, $8, $9, make_interval(secs => $10), $11 ) returning id, queue_name, created_at::timestamp, scheduled_at::timestamp, input, reserved_at::timestamp, started_at::timestamp, cancelled_at::timestamp, finished_at::timestamp, extract(epoch from timeout)::int as timeout, status, output, deadline::timestamp, latest_heartbeat_at::timestamp, failure_reason, error_data, attempt, max_attempts, original_attempt_id, previous_attempt_id, extract(epoch from retry_delay)::int as retry_delay, unique_key; "/utf8>>, _pipe@1 = pog:'query'(_pipe), _pipe@2 = pog:parameter( _pipe@1, pog_ffi:coerce(youid@uuid:to_string(Arg_1)) ), _pipe@3 = pog:parameter(_pipe@2, pog_ffi:coerce(Arg_2)), _pipe@4 = pog:parameter(_pipe@3, pog_ffi:coerce(Arg_3)), _pipe@5 = pog:parameter( _pipe@4, pog_ffi:coerce(gleam@json:to_string(Arg_4)) ), _pipe@6 = pog:parameter(_pipe@5, pog_ffi:coerce(Arg_5)), _pipe@7 = pog:parameter(_pipe@6, pog_ffi:coerce(Arg_6)), _pipe@8 = pog:parameter(_pipe@7, pog_ffi:coerce(Arg_7)), _pipe@9 = pog:parameter( _pipe@8, pog_ffi:coerce(youid@uuid:to_string(Arg_8)) ), _pipe@10 = pog:parameter( _pipe@9, pog_ffi:coerce(youid@uuid:to_string(Arg_9)) ), _pipe@11 = pog:parameter(_pipe@10, pog_ffi:coerce(Arg_10)), _pipe@12 = pog:parameter(_pipe@11, pog_ffi:coerce(Arg_11)), _pipe@13 = pog:returning(_pipe@12, Decoder), pog:execute(_pipe@13, Db). -file("src/m25/internal/sql.gleam", 817). ?DOC(false). -spec reserve_jobs(pog:connection(), binary(), integer()) -> {ok, pog:returned(reserve_jobs_row())} | {error, pog:query_error()}. reserve_jobs(Db, Arg_1, Arg_2) -> Decoder = begin gleam@dynamic@decode:field( 0, uuid_decoder(), fun(Id) -> gleam@dynamic@decode:field( 1, {decoder, fun gleam@dynamic@decode:decode_string/1}, fun(Queue_name) -> gleam@dynamic@decode:field( 2, pog:timestamp_decoder(), fun(Created_at) -> gleam@dynamic@decode:field( 3, pog:timestamp_decoder(), fun(Scheduled_at) -> gleam@dynamic@decode:field( 4, {decoder, fun gleam@dynamic@decode:decode_string/1}, fun(Input) -> gleam@dynamic@decode:field( 5, pog:timestamp_decoder(), fun(Reserved_at) -> gleam@dynamic@decode:field( 6, pog:timestamp_decoder( ), fun(Started_at) -> gleam@dynamic@decode:field( 7, pog:timestamp_decoder( ), fun( Cancelled_at ) -> gleam@dynamic@decode:field( 8, pog:timestamp_decoder( ), fun( Finished_at ) -> gleam@dynamic@decode:field( 9, {decoder, fun gleam@dynamic@decode:decode_int/1}, fun( Timeout ) -> gleam@dynamic@decode:field( 10, {decoder, fun gleam@dynamic@decode:decode_string/1}, fun( Status ) -> gleam@dynamic@decode:field( 11, gleam@dynamic@decode:optional( {decoder, fun gleam@dynamic@decode:decode_string/1} ), fun( Output ) -> gleam@dynamic@decode:field( 12, pog:timestamp_decoder( ), fun( Deadline ) -> gleam@dynamic@decode:field( 13, pog:timestamp_decoder( ), fun( Latest_heartbeat_at ) -> gleam@dynamic@decode:field( 14, gleam@dynamic@decode:optional( {decoder, fun gleam@dynamic@decode:decode_string/1} ), fun( Failure_reason ) -> gleam@dynamic@decode:field( 15, gleam@dynamic@decode:optional( {decoder, fun gleam@dynamic@decode:decode_string/1} ), fun( Error_data ) -> gleam@dynamic@decode:field( 16, {decoder, fun gleam@dynamic@decode:decode_int/1}, fun( Attempt ) -> gleam@dynamic@decode:field( 17, {decoder, fun gleam@dynamic@decode:decode_int/1}, fun( Max_attempts ) -> gleam@dynamic@decode:field( 18, gleam@dynamic@decode:optional( uuid_decoder( ) ), fun( Original_attempt_id ) -> gleam@dynamic@decode:field( 19, gleam@dynamic@decode:optional( uuid_decoder( ) ), fun( Previous_attempt_id ) -> gleam@dynamic@decode:field( 20, {decoder, fun gleam@dynamic@decode:decode_int/1}, fun( Retry_delay ) -> gleam@dynamic@decode:field( 21, gleam@dynamic@decode:optional( {decoder, fun gleam@dynamic@decode:decode_string/1} ), fun( Unique_key ) -> gleam@dynamic@decode:success( {reserve_jobs_row, Id, Queue_name, Created_at, Scheduled_at, Input, Reserved_at, Started_at, Cancelled_at, Finished_at, Timeout, Status, Output, Deadline, Latest_heartbeat_at, Failure_reason, Error_data, Attempt, Max_attempts, Original_attempt_id, Previous_attempt_id, Retry_delay, Unique_key} ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end, _pipe = <<"update m25.job set reserved_at = now() where id in ( select id from m25.job where queue_name = $1 and status = 'pending' and (scheduled_at <= now() or scheduled_at is null) order by created_at limit $2 for update skip locked ) returning id, queue_name, created_at::timestamp, scheduled_at::timestamp, input, reserved_at::timestamp, started_at::timestamp, cancelled_at::timestamp, finished_at::timestamp, extract(epoch from timeout)::int as timeout, status, output, deadline::timestamp, latest_heartbeat_at::timestamp, failure_reason, error_data, attempt, max_attempts, original_attempt_id, previous_attempt_id, extract(epoch from retry_delay)::int as retry_delay, unique_key; "/utf8>>, _pipe@1 = pog:'query'(_pipe), _pipe@2 = pog:parameter(_pipe@1, pog_ffi:coerce(Arg_1)), _pipe@3 = pog:parameter(_pipe@2, pog_ffi:coerce(Arg_2)), _pipe@4 = pog:returning(_pipe@3, Decoder), pog:execute(_pipe@4, Db). -file("src/m25/internal/sql.gleam", 1005). ?DOC(false). -spec time_out_jobs(pog:connection(), binary()) -> {ok, pog:returned(time_out_jobs_row())} | {error, pog:query_error()}. time_out_jobs(Db, Arg_1) -> Decoder = begin gleam@dynamic@decode:field( 0, uuid_decoder(), fun(Id) -> gleam@dynamic@decode:field( 1, {decoder, fun gleam@dynamic@decode:decode_string/1}, fun(Status) -> gleam@dynamic@decode:field( 2, {decoder, fun gleam@dynamic@decode:decode_string/1}, fun(Input) -> gleam@dynamic@decode:field( 3, {decoder, fun gleam@dynamic@decode:decode_int/1}, fun(Attempt) -> gleam@dynamic@decode:field( 4, {decoder, fun gleam@dynamic@decode:decode_int/1}, fun(Max_attempts) -> gleam@dynamic@decode:field( 5, gleam@dynamic@decode:optional( uuid_decoder() ), fun(Original_attempt_id) -> gleam@dynamic@decode:field( 6, gleam@dynamic@decode:optional( uuid_decoder() ), fun( Previous_attempt_id ) -> gleam@dynamic@decode:field( 7, {decoder, fun gleam@dynamic@decode:decode_int/1}, fun( Retry_delay ) -> gleam@dynamic@decode:field( 8, gleam@dynamic@decode:optional( {decoder, fun gleam@dynamic@decode:decode_string/1} ), fun( Unique_key ) -> gleam@dynamic@decode:success( {time_out_jobs_row, Id, Status, Input, Attempt, Max_attempts, Original_attempt_id, Previous_attempt_id, Retry_delay, Unique_key} ) end ) end ) end ) end ) end ) end ) end ) end ) end ) end, _pipe = <<"update m25.job set failure_reason = 'job_timeout', finished_at = now() where queue_name = $1 and status = 'executing' and now() > deadline returning id, status, input, attempt, max_attempts, original_attempt_id, previous_attempt_id, -- TODO: use duration once supported in Squirrel extract(epoch from retry_delay)::int as retry_delay, unique_key; "/utf8>>, _pipe@1 = pog:'query'(_pipe), _pipe@2 = pog:parameter(_pipe@1, pog_ffi:coerce(Arg_1)), _pipe@3 = pog:returning(_pipe@2, Decoder), pog:execute(_pipe@3, Db).